diff --git a/EvoScientist/channels/middleware.py b/EvoScientist/channels/middleware.py index 47de4fc..ba8b60e 100644 --- a/EvoScientist/channels/middleware.py +++ b/EvoScientist/channels/middleware.py @@ -75,11 +75,13 @@ class DedupCache: max_size: int = _DEDUP_MAX, trim_to: int = _DEDUP_TRIM, ttl_seconds: float = _DEDUP_TTL, + clock: Callable[[], float] | None = None, ) -> None: self._seen: OrderedDict[str, float] = OrderedDict() self._max = max_size self._trim = trim_to self._ttl = ttl_seconds + self._clock = clock or time.monotonic # ── public API ────────────────────────────────────────────────── @@ -93,15 +95,16 @@ class DedupCache: if not msg_id: return False - self._prune() + now = self._clock() + self._prune(now) if msg_id in self._seen: # LRU: refresh position and timestamp self._seen.move_to_end(msg_id) - self._seen[msg_id] = time.monotonic() + self._seen[msg_id] = now return True - self._seen[msg_id] = time.monotonic() + self._seen[msg_id] = now if len(self._seen) > self._max: while len(self._seen) > self._trim: self._seen.popitem(last=False) @@ -118,9 +121,9 @@ class DedupCache: # ── internal ──────────────────────────────────────────────────── - def _prune(self) -> None: + def _prune(self, now: float | None = None) -> None: """Remove entries older than *ttl_seconds*.""" - cutoff = time.monotonic() - self._ttl + cutoff = (self._clock() if now is None else now) - self._ttl # OrderedDict is insertion-ordered; oldest entries are first. while self._seen: _key, ts = next(iter(self._seen.items())) @@ -428,11 +431,13 @@ class DedupMiddleware(InboundMiddleware): max_size: int = 1000, trim_to: int = 500, ttl_seconds: float = 3600.0, + clock: Callable[[], float] | None = None, ) -> None: self._cache = DedupCache( max_size=max_size, trim_to=trim_to, ttl_seconds=ttl_seconds, + clock=clock, ) async def process_inbound( diff --git a/tests/test_agent_loader.py b/tests/test_agent_loader.py index fdb84fd..e1293ae 100644 --- a/tests/test_agent_loader.py +++ b/tests/test_agent_loader.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +import threading import pytest @@ -93,6 +94,32 @@ def _make_loader_fn(agent_value="AGENT", fail_with=None, capture=None): return _loader +class _GatedThreadLoader: + """Callable loader that blocks until tests explicitly release it.""" + + def __init__(self, agent_value="AGENT", progress_events=()): + self.agent_value = agent_value + self.progress_events = tuple(progress_events) + self.started = threading.Event() + self.release = threading.Event() + self.finished = threading.Event() + + def __call__(self, *, on_mcp_progress=None): + self.started.set() + self.release.wait(timeout=1) + try: + if on_mcp_progress is not None: + for event in self.progress_events: + on_mcp_progress(*event) + return self.agent_value + finally: + self.finished.set() + + +async def _wait_for_event(event, timeout=1): + return await asyncio.to_thread(event.wait, timeout) + + def _run(coro): return asyncio.run(coro) @@ -125,23 +152,24 @@ class TestBackgroundAgentLoaderStart: _run(_go()) def test_start_cancels_in_flight_prior_task(self): - import time - - def _blocking(*, on_mcp_progress=None): - time.sleep(0.05) - return "LATE" + blocking = _GatedThreadLoader("LATE") async def _go(): - loader = BackgroundAgentLoader(_blocking) + loader = BackgroundAgentLoader(blocking) loader.start() first_task = loader.task + assert first_task is not None + assert await _wait_for_event(blocking.started) # Supersede immediately; asyncio.to_thread wrapper gets cancelled. loader._loader_fn = _make_loader_fn("FRESH") loader.start() agent = await loader.await_ready() assert agent == "FRESH" - # Let the first thread drain so its done callback (gated) fires. - await asyncio.sleep(0.1) + blocking.release.set() + try: + await first_task + except asyncio.CancelledError: + pass assert first_task.cancelled() or first_task.done() _run(_go()) @@ -164,17 +192,11 @@ class TestBackgroundAgentLoaderCallbacks: def test_stale_progress_events_are_dropped(self): """A progress event fired after a newer `start` must not reach the hook.""" - import time - + slow_loader = _GatedThreadLoader( + "slow-agent", progress_events=[("success", "from-slow", "1")] + ) seen: list[str] = [] - # Loader 1 sleeps so its progress event fires AFTER load 2 starts. - def slow_loader(*, on_mcp_progress=None): - time.sleep(0.08) - if on_mcp_progress is not None: - on_mcp_progress("success", "from-slow", "1") - return "slow-agent" - def fast_loader(*, on_mcp_progress=None): if on_mcp_progress is not None: on_mcp_progress("success", "from-fast", "1") @@ -186,13 +208,13 @@ class TestBackgroundAgentLoaderCallbacks: async def _go(): loader.start() - # Supersede before the slow thread's event fires. - await asyncio.sleep(0.01) + assert await _wait_for_event(slow_loader.started) + # Loader 1 waits so its progress event fires AFTER load 2 starts. loader._loader_fn = fast_loader loader.start() await loader.await_ready() - # Let the superseded thread finish (its event is gated out). - await asyncio.sleep(0.1) + slow_loader.release.set() + assert await _wait_for_event(slow_loader.finished) _run(_go()) assert "from-fast" in seen @@ -307,21 +329,17 @@ class TestBackgroundAgentLoaderAdopt: def test_adopt_supersedes_in_flight_load(self): """A late background completion must not overwrite an adopted agent.""" - import time + slow_loader = _GatedThreadLoader("FROM_BACKGROUND") - def _slow(*, on_mcp_progress=None): - time.sleep(0.08) - return "FROM_BACKGROUND" - - loader = BackgroundAgentLoader(_slow) + loader = BackgroundAgentLoader(slow_loader) async def _go(): loader.start() - await asyncio.sleep(0.01) + assert await _wait_for_event(slow_loader.started) loader.adopt("FROM_MODEL") - # Give the background thread time to finish and fire its - # done-callback; the generation token should make it a no-op. - await asyncio.sleep(0.1) + slow_loader.release.set() + assert await _wait_for_event(slow_loader.finished) + await asyncio.sleep(0) assert loader.agent == "FROM_MODEL" _run(_go()) @@ -343,17 +361,15 @@ class TestBackgroundAgentLoaderIsPending: assert not loader.is_pending def test_true_between_start_and_completion(self): - import time + wait_loader = _GatedThreadLoader("ok") - def _wait_loader(*, on_mcp_progress=None): - time.sleep(0.05) - return "ok" - - loader = BackgroundAgentLoader(_wait_loader) + loader = BackgroundAgentLoader(wait_loader) async def _go(): loader.start() + assert await _wait_for_event(wait_loader.started) assert loader.is_pending + wait_loader.release.set() await loader.await_ready() assert not loader.is_pending diff --git a/tests/test_background.py b/tests/test_background.py index d7716f8..1bc99cc 100644 --- a/tests/test_background.py +++ b/tests/test_background.py @@ -91,7 +91,7 @@ def test_stop_already_finished_is_graceful(tmp_path): assert "already finished" in bg.stop(pid) -def test_exited_elapsed_is_frozen(tmp_path): +def test_exited_elapsed_is_frozen(tmp_path, monkeypatch): """Elapsed for an exited process freezes at its runtime, it must not keep growing.""" pid = bg.launch(_true_cmd(), str(tmp_path)) assert _wait_until(lambda: bg._PROCESSES[pid].finished_ts is not None) @@ -99,7 +99,7 @@ def test_exited_elapsed_is_frozen(tmp_path): proc = bg._PROCESSES[pid] assert proc.finished_ts is not None first = bg._elapsed(proc) - time.sleep(1.1) # intentional: prove elapsed stays frozen, not ticking up + monkeypatch.setattr(bg.time, "time", lambda: proc.finished_ts + 100.0) assert bg._elapsed(proc) == first diff --git a/tests/test_channel_comprehensive.py b/tests/test_channel_comprehensive.py index c22d941..56a1624 100644 --- a/tests/test_channel_comprehensive.py +++ b/tests/test_channel_comprehensive.py @@ -15,7 +15,6 @@ Test groups: from __future__ import annotations import asyncio -import time from datetime import datetime from unittest.mock import AsyncMock, MagicMock @@ -48,6 +47,34 @@ from tests.conftest import run_async as _run from tests.fakes import FakeChannelConfig as _FakeConfig from tests.fakes import FakeGraphGateway, StubChannel + +class ManualClock: + def __init__(self) -> None: + self._now = 0.0 + + def __call__(self) -> float: + return self._now + + def advance(self, seconds: float) -> None: + self._now += seconds + + +async def _flush_debounce(ch: StubChannel, sender: str) -> None: + task = ch._debounce_tasks.get(sender) + if task is not None: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + await ch._process_buffered_messages(sender) + + +async def _wait_for_async(predicate) -> None: + while not predicate(): + await asyncio.sleep(0) + + # ═══════════════════════════════════════════════════════════════════ # 1. DedupCache # ═══════════════════════════════════════════════════════════════════ @@ -69,9 +96,10 @@ class TestDedupCache: assert dc.is_duplicate("") is False def test_ttl_expiry(self): - dc = DedupCache(ttl_seconds=0.05) + clock = ManualClock() + dc = DedupCache(ttl_seconds=0.05, clock=clock) dc.is_duplicate("msg_001") - time.sleep(0.1) + clock.advance(0.051) # After TTL, the entry should be pruned assert dc.is_duplicate("msg_001") is False @@ -759,7 +787,7 @@ class TestChannelDebounce: metadata={"chat_id": "c1"}, ) await ch.queue_message(msg) - await asyncio.sleep(0.2) + await _flush_debounce(ch, "u1") # Check bus received the message assert bus.inbound.qsize() == 1 @@ -788,9 +816,8 @@ class TestChannelDebounce: metadata={"chat_id": "c1"}, ) await ch.queue_message(msg) - await asyncio.sleep(0.01) - await asyncio.sleep(0.5) + await _flush_debounce(ch, "u1") assert bus.inbound.qsize() == 1 received = await bus.consume_inbound() assert "part0" in received.content @@ -845,9 +872,8 @@ class TestChannelDebounce: metadata={"chat_id": "c2", "key": "val2"}, ) await ch.queue_message(msg1) - await asyncio.sleep(0.01) await ch.queue_message(msg2) - await asyncio.sleep(0.3) + await _flush_debounce(ch, "u1") received = await bus.consume_inbound() # BUG: metadata is from msg1 only; msg2's metadata is lost @@ -862,7 +888,7 @@ class TestChannelTyping: ch = StubChannel() await ch.start_typing("c1") assert "c1" in ch._typing_tasks - await asyncio.sleep(0.1) + await asyncio.sleep(0) await ch.stop_typing("c1") assert "c1" not in ch._typing_tasks @@ -1030,7 +1056,14 @@ class TestChannelManagerDispatch: ch = StubChannel() # Override send to track calls sent = [] - ch.send = AsyncMock(side_effect=lambda m: sent.append(m) or True) + sent_event = asyncio.Event() + + async def send(msg): + sent.append(msg) + sent_event.set() + return True + + ch.send = send mgr.register(ch) task = asyncio.create_task(mgr._dispatch_outbound()) @@ -1041,7 +1074,7 @@ class TestChannelManagerDispatch: content="hello", ) ) - await asyncio.sleep(0.1) + await asyncio.wait_for(sent_event.wait(), timeout=1.0) task.cancel() try: await task @@ -1068,7 +1101,10 @@ class TestChannelManagerDispatch: content="hello", ) ) - await asyncio.sleep(0.1) + await asyncio.wait_for( + _wait_for_async(lambda: bus.outbound_size == 0), + timeout=1.0, + ) task.cancel() try: await task @@ -1085,8 +1121,10 @@ class TestChannelManagerDispatch: bus = MessageBus() mgr = ChannelManager(bus) ch = StubChannel() + failed_event = asyncio.Event() async def failing_send(msg): + failed_event.set() return False # Indicates failure ch.send = failing_send @@ -1100,7 +1138,7 @@ class TestChannelManagerDispatch: content="hello", ) ) - await asyncio.sleep(0.1) + await asyncio.wait_for(failed_event.wait(), timeout=1.0) task.cancel() try: await task @@ -1121,7 +1159,13 @@ class TestChannelManagerDispatch: bus = MessageBus() mgr = ChannelManager(bus) ch = StubChannel() - ch.send_media = AsyncMock(return_value=False) + failed_event = asyncio.Event() + + async def failing_send_media(**kwargs): + failed_event.set() + return False + + ch.send_media = failing_send_media mgr.register(ch) task = asyncio.create_task(mgr._dispatch_outbound()) @@ -1133,7 +1177,7 @@ class TestChannelManagerDispatch: media=["/tmp/file.png"], ) ) - await asyncio.sleep(0.1) + await asyncio.wait_for(failed_event.wait(), timeout=1.0) task.cancel() try: await task @@ -1444,8 +1488,11 @@ class TestInboundConsumer: consumer = self._make_consumer() # Start and immediately stop task = asyncio.create_task(consumer.run()) - await asyncio.sleep(0.1) + await asyncio.sleep(0) await consumer.stop() + await consumer.bus.publish_inbound( + BusInbound(channel="stub", sender_id="u1", chat_id="c1", content="wake") + ) await task assert consumer._stopping is True @@ -1593,7 +1640,7 @@ class TestIntegration: # Now simulate the bus path via queue_message await ch.queue_message(inbound) - await asyncio.sleep(0.2) + await _flush_debounce(ch, "user1") assert bus.inbound_size == 1 _run(_test()) @@ -1606,9 +1653,14 @@ class TestIntegration: mgr = ChannelManager(bus) ch = StubChannel() media_sent = [] - ch.send_media = AsyncMock( - side_effect=lambda **kw: media_sent.append(kw) or True, - ) + media_event = asyncio.Event() + + async def send_media(**kw): + media_sent.append(kw) + media_event.set() + return True + + ch.send_media = send_media ch.send = AsyncMock(return_value=True) mgr.register(ch) @@ -1621,7 +1673,7 @@ class TestIntegration: media=["/path/doc.pdf"], ) ) - await asyncio.sleep(0.1) + await asyncio.wait_for(media_event.wait(), timeout=1.0) task.cancel() try: await task