diff --git a/tests/test_bulkhead.py b/tests/test_bulkhead.py
index b8e8d2c..ac4c13c 100644
--- a/tests/test_bulkhead.py
+++ b/tests/test_bulkhead.py
@@ -5,7 +5,6 @@
 """
 
 import asyncio
-import contextlib
 import logging
 from collections.abc import Callable, Coroutine
 from http import HTTPStatus
@@ -48,6 +47,29 @@ async def __call__(self, request: httpx2.Request) -> httpx2.Response:
             self.in_flight -= 1
 
 
+class _GatedHandler:
+    """Holds the first `capacity` requests until `release` is set; later requests return at once."""
+
+    def __init__(self, capacity: int) -> None:
+        self.capacity = capacity
+        self.entered = 0
+        self.full = asyncio.Event()
+        self.release = asyncio.Event()
+
+    async def __call__(self, request: httpx2.Request) -> httpx2.Response:
+        self.entered += 1
+        if self.entered == self.capacity:
+            self.full.set()
+        if self.entered <= self.capacity:
+            await self.release.wait()
+        return httpx2.Response(HTTPStatus.OK, request=request)
+
+    async def wait_full(self) -> None:
+        """Wait until `capacity` requests are inside the handler."""
+        async with asyncio.timeout(5.0):
+            await self.full.wait()
+
+
 def _client(
     handler: Callable[[httpx2.Request], httpx2.Response]
     | Callable[[httpx2.Request], Coroutine[Any, Any, httpx2.Response]],
@@ -150,16 +172,12 @@ async def test_max_concurrent_2_observes_at_most_2_in_flight() -> None:
 
 async def test_raises_bulkhead_full_error_when_acquire_timeout_exceeded() -> None:
     """Slot is held by a slow request; a second request with a tiny timeout raises BulkheadFullError."""
-    handler = _SlowHandler(delay=1.0)
+    handler = _GatedHandler(capacity=_MAX_CONCURRENT_1)
     bulkhead = AsyncBulkhead(max_concurrent=_MAX_CONCURRENT_1, acquire_timeout=_ACQUIRE_TIMEOUT_FAST)
     client = _client(handler, bulkhead=bulkhead)
 
-    async def _hold_slot() -> None:
-        await client.get("https://example.test/slow")
-
-    task = asyncio.create_task(_hold_slot())
-    # Yield to let the slow request acquire the semaphore.
-    await asyncio.sleep(0)
+    holder = asyncio.create_task(client.get("https://example.test/slow"))
+    await handler.wait_full()
 
     with pytest.raises(BulkheadFullError) as exc_info:
         await client.get("https://example.test/fast")
@@ -167,34 +185,26 @@ async def _hold_slot() -> None:
     assert exc_info.value.max_concurrent == _MAX_CONCURRENT_1
     assert exc_info.value.acquire_timeout == _ACQUIRE_TIMEOUT_FAST
 
-    # Cancel the lingering slow task to avoid polluting the event loop.
-    task.cancel()
-    with contextlib.suppress(asyncio.CancelledError):
-        await task
+    handler.release.set()
+    await holder
 
 
 async def test_bulkhead_full_error_chains_from_timeout() -> None:
     """BulkheadFullError raised on the async timeout path chains from the TimeoutError."""
-    handler = _SlowHandler(delay=1.0)
+    handler = _GatedHandler(capacity=_MAX_CONCURRENT_1)
     bulkhead = AsyncBulkhead(max_concurrent=_MAX_CONCURRENT_1, acquire_timeout=_ACQUIRE_TIMEOUT_FAST)
     client = _client(handler, bulkhead=bulkhead)
 
-    async def _hold_slot() -> None:
-        await client.get("https://example.test/slow")
-
-    task = asyncio.create_task(_hold_slot())
-    # Yield to let the slow request acquire the semaphore.
-    await asyncio.sleep(0)
+    holder = asyncio.create_task(client.get("https://example.test/slow"))
+    await handler.wait_full()
 
     with pytest.raises(BulkheadFullError) as exc_info:
         await client.get("https://example.test/fast")
 
     assert isinstance(exc_info.value.__cause__, TimeoutError)
 
-    # Cancel the lingering slow task to avoid polluting the event loop.
-    task.cancel()
-    with contextlib.suppress(asyncio.CancelledError):
-        await task
+    handler.release.set()
+    await holder
 
 
 async def test_bounded_wait_raises_bulkhead_full_error() -> None:
@@ -205,34 +215,36 @@ async def test_bounded_wait_raises_bulkhead_full_error() -> None:
     BulkheadFullError fields (max_concurrent / acquire_timeout) carry the
     configured values.
     """
-    handler = _SlowHandler(delay=_ACQUIRE_TIMEOUT_LONG)  # holds slot for 100ms
+    handler = _GatedHandler(capacity=_MAX_CONCURRENT_1)
     client = _client(
         handler,
         bulkhead=AsyncBulkhead(max_concurrent=_MAX_CONCURRENT_1, acquire_timeout=_ACQUIRE_TIMEOUT_SHORT),
     )
 
     first = asyncio.create_task(client.get("https://example.test/a"))
-    await asyncio.sleep(0.005)  # let first acquire the slot
+    await handler.wait_full()
     with pytest.raises(BulkheadFullError) as info:
         await client.get("https://example.test/b")
     assert info.value.max_concurrent == _MAX_CONCURRENT_1
     assert info.value.acquire_timeout == _ACQUIRE_TIMEOUT_SHORT
-    await first  # cleanup
+    handler.release.set()
+    await first
 
 
 async def test_acquire_timeout_zero_fails_fast() -> None:
     """With acquire_timeout=0, the second call raises immediately without waiting."""
-    handler = _SlowHandler(delay=_ACQUIRE_TIMEOUT_LONG)
+    handler = _GatedHandler(capacity=_MAX_CONCURRENT_1)
     client = _client(
         handler,
         bulkhead=AsyncBulkhead(max_concurrent=_MAX_CONCURRENT_1, acquire_timeout=0),
     )
 
     first = asyncio.create_task(client.get("https://example.test/a"))
-    await asyncio.sleep(0.005)
+    await handler.wait_full()
     with pytest.raises(BulkheadFullError) as info:
         await client.get("https://example.test/b")
     assert info.value.acquire_timeout == 0
+    handler.release.set()
     await first
 
 
@@ -276,18 +288,17 @@ def handler(request: httpx2.Request) -> httpx2.Response:
 
 async def test_slot_released_on_cancellation() -> None:
     """If the calling task is cancelled while next() runs, the slot is released."""
-    handler = _SlowHandler(delay=0.5)  # would block indefinitely
+    handler = _GatedHandler(capacity=_MAX_CONCURRENT_1)
     bulkhead = AsyncBulkhead(max_concurrent=_MAX_CONCURRENT_1, acquire_timeout=0)
     client = _client(handler, bulkhead=bulkhead)
 
     first = asyncio.create_task(client.get("https://example.test/a"))
-    await asyncio.sleep(0.01)  # let first acquire and start sleeping in handler
+    await handler.wait_full()
     first.cancel()
     with pytest.raises(asyncio.CancelledError):
         await first
 
     # Slot must now be released — fail-fast=0 next call proves it.
-    handler.delay = 0.0  # speed up the next request
     response = await client.get("https://example.test/b")
     assert response.status_code == HTTPStatus.OK
 
@@ -300,14 +311,14 @@ async def test_cancellation_before_acquire_does_not_hold_slot() -> None:
     release (it must NOT take the slot the cancelled task was waiting for). And once
     first releases, the fresh request must complete normally.
     """
-    handler = _SlowHandler(delay=0.05)
+    handler = _GatedHandler(capacity=_MAX_CONCURRENT_1)
     bulkhead = AsyncBulkhead(max_concurrent=_MAX_CONCURRENT_1, acquire_timeout=None)
     client = _client(handler, bulkhead=bulkhead)
 
     first = asyncio.create_task(client.get("https://example.test/a"))
-    await asyncio.sleep(0.005)  # first acquires
+    await handler.wait_full()
     second = asyncio.create_task(client.get("https://example.test/b"))  # waits for slot
-    await asyncio.sleep(0.005)  # ensure second is parked on acquire
+    await asyncio.sleep(0)  # ensure second is parked on acquire
     second.cancel()
     with pytest.raises(asyncio.CancelledError):
         await second
@@ -315,10 +326,11 @@ async def test_cancellation_before_acquire_does_not_hold_slot() -> None:
     # Third request issued while first still holds the slot — must not see a phantom
     # free slot left by the cancelled second.
     third = asyncio.create_task(client.get("https://example.test/c"))
+    handler.release.set()
     first_response, third_response = await asyncio.gather(first, third)
     assert first_response.status_code == HTTPStatus.OK
     assert third_response.status_code == HTTPStatus.OK
-    assert handler.calls == 2  # noqa: PLR2004 — first and third reached handler; second never did
+    assert handler.entered == 2  # noqa: PLR2004 — first and third reached handler; second never did
 
 
 # Constructed at module scope on purpose — pins the construct-outside-loop behavior.
@@ -416,7 +428,7 @@ async def _sleep(_: float) -> None:  # don't actually wait between retries
 
 async def test_bulkhead_full_error_is_not_retried_by_retry() -> None:
     """AsyncRetry does NOT retry BulkheadFullError — it's neither a StatusError nor a NetworkError/TimeoutError."""
-    handler = _SlowHandler(delay=0.5)  # holds the slot indefinitely
+    handler = _GatedHandler(capacity=_MAX_CONCURRENT_1)
     bulkhead = AsyncBulkhead(max_concurrent=_MAX_CONCURRENT_1, acquire_timeout=0)
     transport = httpx2.MockTransport(handler)
 
@@ -432,30 +444,23 @@ async def test_bulkhead_full_error_is_not_retried_by_retry() -> None:
         ],
     )
 
-    # Fill the slot with a long-lived task.
     first = asyncio.create_task(client.get("https://example.test/holder"))
-    await asyncio.sleep(0.01)
+    await handler.wait_full()
 
     # Second call hits a full AsyncBulkhead. AsyncRetry must NOT swallow + retry it.
     with pytest.raises(BulkheadFullError):
         await client.get("https://example.test/rejected")
     mock_sleep.assert_not_called()  # AsyncRetry never slept — it didn't try to retry
 
-    # Cleanup.
-    first.cancel()
-    with contextlib.suppress(asyncio.CancelledError):
-        await first
+    handler.release.set()
+    await first
 
 
 async def test_bulkhead_rejected_emits_observability_event(caplog: pytest.LogCaptureFixture) -> None:
     """When the bulkhead rejects a request via acquire_timeout, emit one WARNING on httpware.bulkhead."""
     bulkhead = AsyncBulkhead(max_concurrent=1, acquire_timeout=0.0)
-
-    async def slow_handler(request: httpx2.Request) -> httpx2.Response:
-        await asyncio.sleep(0.05)
-        return httpx2.Response(HTTPStatus.OK, request=request)
-
-    transport = httpx2.MockTransport(slow_handler)
+    handler = _GatedHandler(capacity=1)
+    transport = httpx2.MockTransport(handler)
     client = AsyncClient(
         httpx2_client=httpx2.AsyncClient(transport=transport),
         middleware=[bulkhead],
@@ -464,11 +469,12 @@ async def slow_handler(request: httpx2.Request) -> httpx2.Response:
     async with client:
         # First request occupies the only slot. Second should be rejected immediately.
         first_task = asyncio.create_task(client.get("https://example.test/x"))
-        await asyncio.sleep(0)  # let first_task start and acquire the slot
+        await handler.wait_full()
 
         with caplog.at_level(logging.WARNING, logger="httpware.bulkhead"), pytest.raises(BulkheadFullError):
             await client.get("https://example.test/y")
 
+        handler.release.set()
         await first_task
 
     bulkhead_records = [r for r in caplog.records if r.name == "httpware.bulkhead"]
diff --git a/tests/test_bulkhead_props.py b/tests/test_bulkhead_props.py
index 28f528f..6524407 100644
--- a/tests/test_bulkhead_props.py
+++ b/tests/test_bulkhead_props.py
@@ -40,6 +40,29 @@ async def __call__(self, request: httpx2.Request) -> httpx2.Response:
             self.in_flight -= 1
 
 
+class _GatedHandler:
+    """Holds the first `capacity` requests until `release` is set; later requests return at once."""
+
+    def __init__(self, capacity: int) -> None:
+        self.capacity = capacity
+        self.entered = 0
+        self.full = asyncio.Event()
+        self.release = asyncio.Event()
+
+    async def __call__(self, request: httpx2.Request) -> httpx2.Response:
+        self.entered += 1
+        if self.entered == self.capacity:
+            self.full.set()
+        if self.entered <= self.capacity:
+            await self.release.wait()
+        return httpx2.Response(HTTPStatus.OK, request=request)
+
+    async def wait_full(self) -> None:
+        """Wait until `capacity` requests are inside the handler."""
+        async with asyncio.timeout(5.0):
+            await self.full.wait()
+
+
 @given(
     max_concurrent=st.integers(min_value=1, max_value=8),
     n_requests=st.integers(min_value=1, max_value=32),
@@ -71,24 +94,22 @@ async def test_fail_fast_rejects_when_at_capacity(
     max_concurrent: int,
     extra_requests: int,
 ) -> None:
-    handler = _InFlightHandler(delay=0.05)  # hold slots long enough for fail-fast to fire
+    handler = _GatedHandler(capacity=max_concurrent)
     transport = httpx2.MockTransport(handler)
     client = AsyncClient(
         httpx2_client=httpx2.AsyncClient(transport=transport),
         middleware=[AsyncBulkhead(max_concurrent=max_concurrent, acquire_timeout=0)],
     )
 
-    # Fill the bulkhead with max_concurrent long-running tasks.
     holders = [asyncio.create_task(client.get(f"https://example.test/hold-{i}")) for i in range(max_concurrent)]
-    await asyncio.sleep(0.005)  # let the holders acquire their slots
-
-    # Any extra requests should fail fast with BulkheadFullError.
-    for i in range(extra_requests):
-        with pytest.raises(BulkheadFullError):
-            await client.get(f"https://example.test/extra-{i}")
-
-    # Cleanup the holders.
-    await asyncio.gather(*holders)
+    try:
+        await handler.wait_full()
+        for i in range(extra_requests):
+            with pytest.raises(BulkheadFullError):
+                await client.get(f"https://example.test/extra-{i}")
+    finally:
+        handler.release.set()
+        await asyncio.gather(*holders)
 
 
 @given(
diff --git a/tests/test_bulkhead_sync.py b/tests/test_bulkhead_sync.py
index 213c7a6..f728ee9 100644
--- a/tests/test_bulkhead_sync.py
+++ b/tests/test_bulkhead_sync.py
@@ -49,6 +49,27 @@ def __call__(self, request: httpx2.Request) -> httpx2.Response:
                 self.in_flight -= 1
 
 
+class _GatedHandler:
+    """Holds the first `capacity` requests until `release` is set; later requests return at once."""
+
+    def __init__(self, capacity: int) -> None:
+        self.capacity = capacity
+        self.lock = threading.Lock()
+        self.entered = 0
+        self.full = threading.Event()
+        self.release = threading.Event()
+
+    def __call__(self, request: httpx2.Request) -> httpx2.Response:
+        with self.lock:
+            self.entered += 1
+            entered = self.entered
+        if entered == self.capacity:
+            self.full.set()
+        if entered <= self.capacity:
+            self.release.wait()
+        return httpx2.Response(HTTPStatus.OK, request=request)
+
+
 def _client(
     handler: Callable[[httpx2.Request], httpx2.Response],
     *,
@@ -110,7 +131,7 @@ def test_serializes_at_capacity() -> None:
 
 
 def test_acquire_timeout_rejects_when_no_slot_available() -> None:
-    handler = _SlowHandler(delay=0.1)
+    handler = _GatedHandler(capacity=_MAX_CONCURRENT_1)
     client = _client(
         handler,
         bulkhead=Bulkhead(max_concurrent=_MAX_CONCURRENT_1, acquire_timeout=_ACQUIRE_TIMEOUT_FAST),
@@ -118,20 +139,20 @@ def test_acquire_timeout_rejects_when_no_slot_available() -> None:
 
     holder = threading.Thread(target=client.get, args=("https://example.test/hold",))
     holder.start()
-    # Give the holder time to acquire the only slot
-    time.sleep(0.01)
     try:
+        assert handler.full.wait(timeout=5.0)
         with pytest.raises(BulkheadFullError) as info:
             client.get("https://example.test/blocked")
         assert info.value.max_concurrent == _MAX_CONCURRENT_1
         assert info.value.acquire_timeout == _ACQUIRE_TIMEOUT_FAST
     finally:
+        handler.release.set()
         holder.join()
 
 
 def test_bulkhead_full_error_no_chaining() -> None:
     """BulkheadFullError raised on the sync timeout path has no __cause__ (no active exception)."""
-    handler = _SlowHandler(delay=0.1)
+    handler = _GatedHandler(capacity=_MAX_CONCURRENT_1)
     client = _client(
         handler,
         bulkhead=Bulkhead(max_concurrent=_MAX_CONCURRENT_1, acquire_timeout=_ACQUIRE_TIMEOUT_FAST),
@@ -139,13 +160,13 @@ def test_bulkhead_full_error_no_chaining() -> None:
 
     holder = threading.Thread(target=client.get, args=("https://example.test/hold",))
     holder.start()
-    # Give the holder time to acquire the only slot
-    time.sleep(0.01)
     try:
+        assert handler.full.wait(timeout=5.0)
         with pytest.raises(BulkheadFullError) as exc_info:
             client.get("https://example.test/blocked")
         assert exc_info.value.__cause__ is None
     finally:
+        handler.release.set()
         holder.join()
 
 
@@ -171,19 +192,20 @@ def boom(request: httpx2.Request) -> httpx2.Response:  # noqa: ARG001
 
 
 def test_emits_rejected_event(caplog: pytest.LogCaptureFixture) -> None:
-    handler = _SlowHandler(delay=0.1)
+    handler = _GatedHandler(capacity=_MAX_CONCURRENT_1)
     client = _client(
         handler,
         bulkhead=Bulkhead(max_concurrent=_MAX_CONCURRENT_1, acquire_timeout=_ACQUIRE_TIMEOUT_FAST),
     )
     holder = threading.Thread(target=client.get, args=("https://example.test/hold",))
     holder.start()
-    time.sleep(0.01)
     try:
+        assert handler.full.wait(timeout=5.0)
         with caplog.at_level(logging.WARNING, logger="httpware.bulkhead"), pytest.raises(BulkheadFullError):
             client.get("https://example.test/blocked")
         assert any("bulkhead rejected" in r.getMessage() for r in caplog.records)
     finally:
+        handler.release.set()
         holder.join()
 
 
diff --git a/tests/test_bulkhead_sync_props.py b/tests/test_bulkhead_sync_props.py
index bc366eb..6c23365 100644
--- a/tests/test_bulkhead_sync_props.py
+++ b/tests/test_bulkhead_sync_props.py
@@ -74,16 +74,16 @@ def test_in_flight_never_exceeds_max_concurrent(
 
 
 class _BarrierHandler:
-    """Handler that signals slot acquisition via a threading.Barrier before holding the slot."""
+    """Handler that signals slot acquisition via a threading.Barrier, then holds the slot until `release` is set."""
 
-    def __init__(self, barrier: threading.Barrier) -> None:
+    def __init__(self, barrier: threading.Barrier, release: threading.Event) -> None:
         self._barrier = barrier
+        self._release = release
 
     def __call__(self, request: httpx2.Request) -> httpx2.Response:
         # Signal that this holder has acquired a bulkhead slot and is now in-flight.
         self._barrier.wait(timeout=5.0)
-        # Hold the slot long enough for the over-limit requests to be rejected.
-        time.sleep(0.05)
+        self._release.wait()
         return httpx2.Response(HTTPStatus.OK, request=request)
 
 
@@ -100,7 +100,8 @@ def test_fail_fast_rejects_when_at_capacity(
     # holder has acquired its bulkhead slot (i.e. is inside the handler).
     # timeout=5.0 sets the default for all barrier.wait() calls.
     acquired_barrier = threading.Barrier(max_concurrent + 1, timeout=5.0)
-    handler = _BarrierHandler(acquired_barrier)
+    release = threading.Event()
+    handler = _BarrierHandler(acquired_barrier, release)
     transport = httpx2.MockTransport(handler)
     client = Client(
         httpx2_client=httpx2.Client(transport=transport),
@@ -110,18 +111,19 @@ def test_fail_fast_rejects_when_at_capacity(
     # Fill the bulkhead with max_concurrent long-running threads.
     pool = ThreadPoolExecutor(max_workers=max_concurrent + extra_requests)
     holders = [pool.submit(client.get, f"https://example.test/hold-{i}") for i in range(max_concurrent)]
-    # Wait deterministically — barrier releases only once ALL holders are inside the handler.
-    acquired_barrier.wait(timeout=5.0)
-
-    # Any extra requests should fail fast with BulkheadFullError.
-    for i in range(extra_requests):
-        with pytest.raises(BulkheadFullError):
-            client.get(f"https://example.test/extra-{i}")
-
-    # Cleanup the holders.
-    for f in holders:
-        f.result()
-    pool.shutdown()
+    try:
+        # Wait deterministically — barrier releases only once ALL holders are inside the handler.
+        acquired_barrier.wait(timeout=5.0)
+
+        # Any extra requests should fail fast with BulkheadFullError.
+        for i in range(extra_requests):
+            with pytest.raises(BulkheadFullError):
+                client.get(f"https://example.test/extra-{i}")
+    finally:
+        release.set()
+        for f in holders:
+            f.result()
+        pool.shutdown()
 
 
 @given(