diff --git a/tests/agent/test_context_refs_concurrent.py b/tests/agent/test_context_refs_concurrent.py index a4dd64624b4..f1063bdc6b4 100644 --- a/tests/agent/test_context_refs_concurrent.py +++ b/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 diff --git a/tests/agent/test_memory_boundary_commit.py b/tests/agent/test_memory_boundary_commit.py index e2e6aec1b70..482cfadade9 100644 --- a/tests/agent/test_memory_boundary_commit.py +++ b/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}" diff --git a/tests/plugins/memory/test_mem0_v3.py b/tests/plugins/memory/test_mem0_v3.py index f6c28ad84ed..48d66d480b2 100644 --- a/tests/plugins/memory/test_mem0_v3.py +++ b/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 @@ -186,19 +187,47 @@ class TestMem0Prefetch: 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?")