fix: remove remaining flaky wall-clock test assertions

tests/core/tools/adapters/vibe/test_mcp_init_timeout.py

@@ -2,7 +2,6 @@
 agent setup (or pin resources) indefinitely."""
 
 import asyncio
-import time
 
 import pytest
 
@@ -18,6 +17,7 @@
 
 
 @pytest.mark.asyncio
+@pytest.mark.timeout(30)
 async def test_stalled_server_times_out_and_other_servers_still_load(monkeypatch):
     """A server whose initialize/list-tools stalls is skipped at the timeout;
     the remaining servers still load."""
@@ -36,27 +36,23 @@ async def fake_load_direct(server_name, connection, **kwargs):
 
     monkeypatch.setattr(mcp_adapter_module, "_load_direct_mcp_tools", fake_load_direct)
 
-    started = time.monotonic()
     result = await load_mcp_tools_as_agent_tools(
         {
             "stalled": {"transport": "streamable_http", "url": "http://x"},
             "healthy": {"transport": "streamable_http", "url": "http://y"},
         }
     )
-    elapsed = time.monotonic() - started
 
     assert result.tools == (healthy_tool,)
     assert result.loaded_servers == ("healthy",)
     assert len(result.failures) == 1
     assert result.failures[0].server_name == "stalled"
     assert result.failures[0].phase is MCPFailurePhase.INITIALIZE
     assert result.failures[0].error_type == "TimeoutError"
-    # 1s timeout for the stalled server plus fast healthy load; the old
-    # behavior blocked forever.
-    assert elapsed < 5
 
 
 @pytest.mark.asyncio
+@pytest.mark.timeout(30)
 async def test_bounded_load_returns_even_when_cleanup_hangs():
     """The bound must hold even if the load task ignores cancellation (e.g. a
     hung streamable-HTTP session blocking in __aexit__)."""
@@ -80,12 +76,9 @@ async def uncancellable_load():
                     continue
         return []  # pragma: no cover
 
-    started = time.monotonic()
     with pytest.raises(TimeoutError):
         await _load_server_tools_bounded("hung", uncancellable_load(), 1)
-    elapsed = time.monotonic() - started
 
-    assert elapsed < 5
     # The load was cancelled (cleanup began) but the caller did not wait on
     # it: the bounded call returned while cleanup was still blocked.
     await asyncio.wait_for(cleanup_entered.wait(), timeout=5)
@@ -94,6 +87,7 @@ async def uncancellable_load():
 
 
 @pytest.mark.asyncio
+@pytest.mark.timeout(30)
 async def test_burst_larger_than_gate_does_not_fan_out(monkeypatch):
     """A burst of concurrent loads for the same hung server must not create
     more underlying load tasks (transports/sockets) than the per-server cap:
@@ -120,13 +114,9 @@ async def one_caller():
         with pytest.raises(TimeoutError):
             await _load_server_tools_bounded("burst-server", uncancellable_load(), 1)
 
-    began = time.monotonic()
     await asyncio.gather(*(one_caller() for _ in range(6)))
-    elapsed = time.monotonic() - began
 
-    # Every caller returned within its own bound...
-    assert elapsed < 5
-    # ...but only cap-many loads (transports) ever started; the abandoned
+    # Only cap-many loads (transports) ever started; the abandoned
     # ones keep holding their slots so the other four callers failed fast
     # at the gate.
     assert started_loads == 2

tests/core/tools/core/test_document_search_collection_concurrency.py

@@ -359,6 +359,7 @@ async def _list(
 
 
 @pytest.mark.asyncio
+@pytest.mark.timeout(30)
 async def test_a_stuck_collection_times_out_without_holding_the_batch(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
@@ -369,19 +370,15 @@ async def test_a_stuck_collection_times_out_without_holding_the_batch(
 
     def search(*, collection: str, **_kwargs: Any) -> SearchPipelineResult:
         if collection == "stuck":
-            release.wait(10)
+            release.wait()
         return _pipeline_result(collection)
 
     monkeypatch.setattr(document_search, "run_document_search", search)
 
-    started = time.perf_counter()
     try:
         result = await document_search._search_knowledge_base_impl(_args(), user_id=1)
     finally:
         release.set()
-    elapsed = time.perf_counter() - started
-
-    assert elapsed < 5
     assert [entry.collection for entry in result.results] == ["fast"]
     assert "stuck: search timed out after 0.2s" in result.summary
 

tests/web/api/test_kb_collections_listing.py

@@ -5,7 +5,6 @@
 """
 
 import asyncio
-import time
 from types import SimpleNamespace
 
 import pytest
@@ -137,6 +136,7 @@ async def test_team_owner_scan_results_are_merged_per_owner(team_listing_env):
 
 
 @pytest.mark.asyncio
+@pytest.mark.timeout(30)
 async def test_scan_timeout_comes_from_config(monkeypatch):
     """The per-scan deadline must be configurable, not a hardcoded constant."""
     monkeypatch.setenv("XAGENT_KB_COLLECTIONS_TIMEOUT_SECONDS", "1")
@@ -145,16 +145,22 @@ async def test_scan_timeout_comes_from_config(monkeypatch):
     )
 
     async def _slow_scan(user_id: int, is_admin: bool = False):
-        await asyncio.sleep(5)
+        await asyncio.Event().wait()
         return _result("never")
 
     monkeypatch.setattr(kb_api, "list_collections", _slow_scan)
     user = SimpleNamespace(id=1, is_admin=False)
 
-    started = time.perf_counter()
+    configured_timeouts = []
+    original_wait_for = asyncio.wait_for
+
+    async def observed_wait_for(awaitable, timeout):
+        configured_timeouts.append(timeout)
+        return await original_wait_for(awaitable, timeout=timeout)
+
+    monkeypatch.setattr(kb_api.asyncio, "wait_for", observed_wait_for)
     with pytest.raises(kb_api.HTTPException) as excinfo:
         await kb_api.list_collections_api(_user=user, db=None)
-    elapsed = time.perf_counter() - started
 
     assert excinfo.value.status_code == 503
-    assert elapsed < 3, f"configured 1s timeout was not honoured ({elapsed:.3f}s)"
+    assert configured_timeouts == [1]

tests/web/api/test_upload_connection_boundary.py

@@ -6,7 +6,6 @@
 import builtins
 import io
 import threading
-import time
 from pathlib import Path
 
 import pytest
@@ -96,14 +95,18 @@ def get_upload_path(
 class _SlowTrackingSource(io.BytesIO):
     """A synchronous source that reveals an accidental event-loop read."""
 
-    def __init__(self, payload: bytes, delay_seconds: float) -> None:
+    def __init__(self, payload: bytes) -> None:
         super().__init__(payload)
-        self.delay_seconds = delay_seconds
+        self.started = threading.Event()
+        self.release = threading.Event()
+        self.loop_thread = threading.get_ident()
         self.read_threads: list[int] = []
 
     def read(self, size: int = -1) -> bytes:
         self.read_threads.append(threading.get_ident())
-        time.sleep(self.delay_seconds)
+        self.started.set()
+        assert threading.get_ident() != self.loop_thread
+        assert self.release.wait(timeout=30), "upload read was never released"
         return super().read(size)
 
 
@@ -213,14 +216,13 @@ async def test_staging_copy_reads_off_loop_with_no_database_checkout(
         files_api, "get_upload_path", _stage_path_in(upload_root, user_id)
     )
     engine = _install_one_slot_queue_pool(monkeypatch)
-    source = _SlowTrackingSource(b"bounded-copy", delay_seconds=0.2)
+    source = _SlowTrackingSource(b"bounded-copy")
     upload = UploadFile(
         filename="off-loop.txt",
         file=source,
         headers={"content-type": "text/plain"},
     )
     loop_thread = threading.get_ident()
-    started_at = asyncio.get_running_loop().time()
     task = asyncio.create_task(
         files_api.store_uploaded_files(
             upload_items=[upload],
@@ -232,15 +234,20 @@ async def test_staging_copy_reads_off_loop_with_no_database_checkout(
         )
     )
     try:
-        await asyncio.sleep(0.01)
-        assert asyncio.get_running_loop().time() - started_at < 0.1
+        assert await asyncio.to_thread(source.started.wait, 30)
+        assert not task.done()
         assert engine.pool.checkedout() == 0
-        await task
+        source.release.set()
+        await asyncio.wait_for(task, timeout=30)
         assert source.read_threads
         assert all(thread_id != loop_thread for thread_id in source.read_threads)
         assert engine.pool.checkedout() == 0
     finally:
-        engine.dispose()
+        source.release.set()
+        try:
+            await asyncio.wait_for(task, timeout=30)
+        finally:
+            engine.dispose()
 
 
 def test_reserve_and_copy_enforces_max_size_and_cleans_partial_file(

tests/web/api/v1/test_auth.py

@@ -519,15 +519,19 @@ async def test_record_key_usage_pool_wait_keeps_event_loop_responsive(
     async def invoke_usage() -> None:
         await record_key_usage(prefix)
 
-    usage_task = asyncio.create_task(invoke_usage())
-    started_at = asyncio.get_running_loop().time()
-    ticker_task = asyncio.create_task(asyncio.sleep(0.02))
+    from tests.web.pool_contention_shared import GUARD_TIMEOUT, gated_pool_checkout
+
     try:
-        await ticker_task
-        assert asyncio.get_running_loop().time() - started_at < 0.08
+        with gated_pool_checkout(engine) as gate:
+            usage_task = asyncio.create_task(invoke_usage())
+            try:
+                await gate.wait_until_contending()
+                assert not usage_task.done()
+            finally:
+                held_connection.close()
+                gate.let_through()
+                await asyncio.wait_for(usage_task, timeout=GUARD_TIMEOUT)
     finally:
-        held_connection.close()
-        await usage_task
         engine.dispose()
 
 

tests/web/api/v1/test_tasks.py

@@ -17,7 +17,6 @@
 import asyncio
 import io
 import threading
-import time
 from dataclasses import replace
 from datetime import UTC, datetime
 from types import SimpleNamespace
@@ -618,10 +617,15 @@ async def test_upload_durable_phase_releases_pool_and_event_loop(
     engine = _install_one_slot_queue_pool(monkeypatch)
     checked_out_during_durable: list[int] = []
     original_sync = ManagedFileRef.sync_to_durable
+    entered = threading.Event()
+    release = threading.Event()
+    loop_thread = threading.get_ident()
 
     def delayed_sync(self, *args, **kwargs):  # type: ignore[no-untyped-def]
         checked_out_during_durable.append(engine.pool.checkedout())
-        time.sleep(0.1)
+        entered.set()
+        assert threading.get_ident() != loop_thread
+        assert release.wait(timeout=30), "durable upload was never released"
         return original_sync(self, *args, **kwargs)
 
     monkeypatch.setattr(ManagedFileRef, "sync_to_durable", delayed_sync)
@@ -641,17 +645,18 @@ async def upload_once() -> None:
             user_id=user_id,
         )
 
-    started_at = asyncio.get_running_loop().time()
     upload_task = asyncio.create_task(upload_once())
-    ticker_task = asyncio.create_task(asyncio.sleep(0.02))
     try:
-        await ticker_task
-        assert asyncio.get_running_loop().time() - started_at < 0.08
-        await upload_task
+        assert await asyncio.to_thread(entered.wait, 30)
+        assert not upload_task.done()
         assert checked_out_during_durable == [0]
         assert engine.pool.checkedout() == 0
     finally:
-        engine.dispose()
+        release.set()
+        try:
+            await asyncio.wait_for(upload_task, timeout=30)
+        finally:
+            engine.dispose()
 
 
 @pytest.mark.asyncio

tests/web/services/test_task_orchestrator.py

@@ -20,7 +20,7 @@
 from concurrent.futures import ThreadPoolExecutor
 from contextlib import ExitStack, contextmanager
 from datetime import datetime, timedelta, timezone
-from threading import Barrier, get_ident
+from threading import Barrier, Event, get_ident
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
@@ -1411,30 +1411,24 @@ async def test_schedule_claimed_create_turn_offloads_cache_invalidation(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     """A committed domain claim must not invalidate Redis on the event loop."""
-    import time as _time
-
     task_id = 987654
     event_loop_thread = get_ident()
     invalidations: list[tuple[int, int]] = []
-    ticker_stop = asyncio.Event()
-    ticks = 0
+    entered = Event()
+    release = Event()
 
     def slow_invalidate(observed_task_id: int) -> None:
         invalidations.append((observed_task_id, get_ident()))
-        _time.sleep(0.08)
+        entered.set()
+        assert get_ident() != event_loop_thread
+        assert release.wait(timeout=30), "cache invalidation was never released"
 
     async def fake_schedule(**_kwargs):
         async def done() -> None:
             return None
 
         return asyncio.create_task(done())
 
-    async def ticker() -> None:
-        nonlocal ticks
-        while not ticker_stop.is_set():
-            ticks += 1
-            await asyncio.sleep(0.005)
-
     monkeypatch.setattr(
         task_orchestrator_module,
         "invalidate_task_cache",
@@ -1458,24 +1452,26 @@ async def ticker() -> None:
         run_id="committed-run",
     )
 
-    ticker_task = asyncio.create_task(ticker())
-    try:
-        started = await TaskTurnOrchestrator.schedule_claimed_create_turn(
+    startup = asyncio.create_task(
+        TaskTurnOrchestrator.schedule_claimed_create_turn(
             task_id=task_id,
             task_owner_user_id=1,
             actor_user_id=1,
             payload=TaskTurnPayload("start"),
             claimed=claimed,
         )
-        await started.background_task
+    )
+    try:
+        assert await asyncio.to_thread(entered.wait, 30)
+        assert not startup.done()
     finally:
-        ticker_stop.set()
-        await ticker_task
+        release.set()
+        started = await asyncio.wait_for(startup, timeout=30)
+    await started.background_task
 
     assert len(invalidations) == 1
     assert invalidations[0][0] == task_id
     assert invalidations[0][1] != event_loop_thread
-    assert ticks >= 3, "claim-cache invalidation blocked the asyncio event loop"
 
 
 @pytest.mark.asyncio