test: prove concurrency and non-blocking with barriers/witnesses, not wall-clock bounds

tests/agent/test_context_refs_concurrent.py

@@ -1,15 +1,16 @@
 """Tests for concurrent @-reference expansion in context_references.
 
-RED before the refactor: test_refs_expand_concurrently asserts that N URL refs
-(each a ~0.2s fetch) complete in roughly one fetch-time, not N×. On the serial
-`for ref in refs: await` loop this FAILS (takes ~N×0.2s); after switching to
-asyncio.gather it passes. The output-contract test guards that concurrency does
-NOT change ordering, warnings, blocks, or token accounting.
+test_refs_expand_concurrently asserts that N URL refs are fetched CONCURRENTLY.
+It proves this with an asyncio.Barrier rendezvous rather than a stopwatch: all
+N fetches must be in flight at the same instant before any is allowed to
+return. On the serial `for ref in refs: await` loop the first fetch waits for
+partners that never arrive and the test fails; with asyncio.gather it passes.
+The output-contract test guards that concurrency does NOT change ordering,
+warnings, blocks, or token accounting.
 """
 from __future__ import annotations
 
 import asyncio
-import time
 
 import pytest
 
@@ -26,13 +27,49 @@ async def _slow_fetcher(url: str) -> str:
 async def test_refs_expand_concurrently(tmp_path):
     # Three independent URL refs in one message.
     msg = "see @url:https://a.example/x @url:https://b.example/y @url:https://c.example/z please"
-    t0 = time.perf_counter()
-    res = await preprocess_context_references_async(
-        msg, cwd=tmp_path, context_length=100_000, url_fetcher=_slow_fetcher,
-    )
-    elapsed = time.perf_counter() - t0
-    # Serial would be ~0.6s (3×0.2). Concurrent ~0.2s. Assert well under 2× one fetch.
-    assert elapsed < 0.4, f"expected concurrent (~0.2s), got {elapsed:.2f}s (serial?)"
+
+    # Concurrency is proven by construction, not by measuring elapsed time.
+    #
+    # The old form asserted `elapsed < 0.4` ("well under 2x one 0.2s fetch").
+    # That makes the event-loop scheduler and any fixed setup part of the
+    # assertion: under a loaded CI box the inequality can flip with nothing
+    # wrong in the code under test, and the margin shrinks silently if setup
+    # cost is ever added ahead of dispatch.
+    #
+    # A barrier asserts the invariant directly: all THREE fetches must be
+    # inside the fetcher AT THE SAME TIME before any is allowed to return. If
+    # expansion ever goes serial the first fetch blocks waiting for partners
+    # that will not arrive, the barrier times out, and the test fails with an
+    # explicit message. No wall-clock constant, no load sensitivity.
+    N_REFS = 3
+    rendezvous = asyncio.Barrier(N_REFS)
+    entered: list[str] = []
+    overlapped = asyncio.Event()
+
+    async def barrier_fetcher(url: str) -> str:
+        entered.append(url)
+        # Generous relative to real scheduling latency (a rendezvous between
+        # already-dispatched coroutines needs milliseconds), but finite so a
+        # serial regression fails fast instead of hanging the suite.
+        async with asyncio.timeout(10):
+            await rendezvous.wait()
+        overlapped.set()
+        return f"CONTENT[{url}]"
+
+    try:
+        res = await preprocess_context_references_async(
+            msg, cwd=tmp_path, context_length=100_000, url_fetcher=barrier_fetcher,
+        )
+    except (asyncio.BrokenBarrierError, TimeoutError):  # pragma: no cover - serial regression
+        pytest.fail(
+            "references did not expand concurrently: a fetch reached the "
+            f"rendezvous alone, so expansion never had {N_REFS} fetches in "
+            f"flight at once (entered: {entered})"
+        )
+
+    # The barrier only clears when all three fetches are in flight together.
+    assert overlapped.is_set(), f"references never overlapped (entered: {entered})"
+    assert len(entered) == N_REFS, f"expected {N_REFS} fetches, got {entered}"
     # All three blocks present, in order.
     assert res.expanded
     body = res.message

tests/agent/test_memory_boundary_commit.py

@@ -66,14 +66,27 @@ def test_boundary_commit_delivers_end_strictly_before_switch():
     mm = _make_manager(provider)
 
     msgs = [{"role": "user", "content": "old turn"}]
-    t0 = time.monotonic()
     mm.commit_session_boundary_async(
         msgs, new_session_id="new-sid", parent_session_id="old-sid"
     )
-    # Caller returns immediately — the slow extraction must not block /new.
-    assert time.monotonic() - t0 < 0.1
+    # DETERMINISTIC non-blocking witness — replaces `assert elapsed < 0.1`.
+    #
+    # The old form timed `commit_session_boundary_async` and required it under
+    # 100ms, which makes the scheduler part of the assertion: thread startup
+    # alone can exceed that on a loaded box, flipping the inequality with
+    # nothing wrong in the code under test.
+    #
+    # The real contract is that the caller returns WITHOUT waiting for the slow
+    # extraction. Assert it directly: the background `on_session_end` sleeps
+    # 0.15s before recording anything, so if the caller had blocked on it, the
+    # provider would already have recorded the "end" call by the time we get
+    # here. An empty call list is a positive witness that /new was not gated.
+    assert provider.calls == [], (
+        "commit_session_boundary_async blocked on the slow extraction: "
+        f"provider already recorded {provider.calls} before the caller returned"
+    )
 
-    assert mm.flush_pending(timeout=5)
+    assert mm.flush_pending(timeout=30)
 
     kinds = [c[0] for c in provider.calls]
     assert kinds == ["end", "switch"], f"ordering violated: {provider.calls}"

tests/plugins/memory/test_mem0_v3.py

@@ -1,6 +1,7 @@
 """Tests for Mem0 v3 API — new tool names, paginated responses, update/delete tools."""
 
 import json
+import threading
 import time
 import pytest
 
@@ -313,19 +314,47 @@ def test_on_turn_start_queues_current_query(self):
         assert len([c for c in backend.captured if c[0] == "search"]) == 1
 
     def test_slow_prefetch_returns_quickly(self, monkeypatch):
+        entered = threading.Event()
+        release = threading.Event()
+        search_returned = threading.Event()
+
         class SlowBackend(FakeBackend):
             def search(self, query, *, filters, top_k=10, rerank=True):
-                time.sleep(0.2)
-                return super().search(query, filters=filters, top_k=top_k, rerank=rerank)
+                entered.set()
+                try:
+                    release.wait(30)
+                    return super().search(
+                        query, filters=filters, top_k=top_k, rerank=rerank
+                    )
+                finally:
+                    search_returned.set()
 
         monkeypatch.setattr(mem0_plugin, "_PREFETCH_WAIT_SECS", 0.01)
         provider = self._make_provider(
             SlowBackend(search_results=[{"id": "m1", "memory": "lives in Berlin"}])
         )
-        started = time.monotonic()
+        # DETERMINISTIC non-blocking witness — replaces `assert elapsed < 0.1`.
+        #
+        # The old form slept 0.2s in the backend and asserted prefetch returned
+        # in under 0.1s. That makes the OS scheduler part of the assertion: on
+        # a loaded box thread startup alone can eat the 100ms budget, so the
+        # inequality flips with nothing wrong in the code under test. Observed
+        # failing in a full-directory run of tests/plugins/memory.
+        #
+        # The real contract is that prefetch gives up on the slow backend
+        # instead of waiting for it. Assert it directly: the backend search is
+        # STILL PARKED (release unset, so `search_returned` cannot be set). If
+        # prefetch ever waited for the backend, the search would have returned
+        # first and this fails. No wall-clock constant.
         assert provider.prefetch("where do I live?") == ""
-        assert time.monotonic() - started < 0.1
-        provider._prefetch_thread.join(timeout=1)
+        assert entered.wait(30), "prefetch never reached the backend"
+        assert not search_returned.is_set(), (
+            "prefetch blocked on the slow backend: the backend search had "
+            "already returned by the time prefetch did"
+        )
+
+        release.set()
+        provider._prefetch_thread.join(timeout=30)
         assert "lives in Berlin" in provider.prefetch("where do I live?")
 
     def test_prefetch_empty_results_returns_empty(self):

tests/tools/test_mcp_tool.py

@@ -1647,31 +1647,55 @@ def test_shutdown_is_parallel(self):
         """Multiple servers are shut down in parallel via asyncio.gather."""
         import tools.mcp_tool as mcp_mod
         from tools.mcp_tool import shutdown_mcp_servers, _servers
-        import time
+        import threading
 
         _servers.clear()
 
-        # 3 servers each taking 1s to shut down
-        for i in range(3):
+        # Concurrency is proven by construction, not by measuring elapsed time.
+        #
+        # The old form gave each server a 1s sleep and asserted the total came
+        # in under 2.5s ("parallel: ~1s, not ~3s"). That makes the scheduler
+        # part of the assertion — under a loaded CI box the inequality can flip
+        # with nothing wrong in the code under test — and the margin silently
+        # absorbs any fixed setup added ahead of dispatch.
+        #
+        # A barrier asserts the invariant directly: all THREE shutdowns must be
+        # in flight AT THE SAME TIME before any is allowed to complete. If
+        # shutdown ever goes serial the first one waits for partners that never
+        # arrive, the barrier breaks on its timeout, and the test fails.
+        N_SERVERS = 3
+        rendezvous = threading.Barrier(N_SERVERS)
+        overlapped = threading.Event()
+        entered: list[str] = []
+
+        for i in range(N_SERVERS):
             mock_server = MagicMock()
-            mock_server.name = f"srv_{i}"
-            async def slow_shutdown():
-                await asyncio.sleep(1)
+            name = f"srv_{i}"
+            mock_server.name = name
+
+            async def slow_shutdown(_name=name):
+                entered.append(_name)
+                # Generous vs real scheduling latency, finite so a serial
+                # regression fails fast instead of hanging the suite.
+                await asyncio.to_thread(rendezvous.wait, 10)
+                overlapped.set()
+
             mock_server.shutdown = slow_shutdown
-            _servers[f"srv_{i}"] = mock_server
+            _servers[name] = mock_server
 
         mcp_mod._ensure_mcp_loop()
         try:
-            start = time.monotonic()
             shutdown_mcp_servers()
-            elapsed = time.monotonic() - start
         finally:
             mcp_mod._mcp_loop = None
             mcp_mod._mcp_thread = None
 
         assert len(_servers) == 0
-        # Parallel: ~1s, not ~3s. Allow some margin.
-        assert elapsed < 2.5, f"Shutdown took {elapsed:.1f}s, expected ~1s (parallel)"
+        # The barrier only clears when all three shutdowns are in flight together.
+        assert overlapped.is_set(), (
+            f"server shutdowns never overlapped — ran serially (entered: {entered})"
+        )
+        assert len(entered) == N_SERVERS, f"expected {N_SERVERS} shutdowns, got {entered}"
 
 
 # ---------------------------------------------------------------------------