feat(bulkhead): detect-and-raise on cross-event-loop AsyncBulkhead reuse

AsyncBulkhead's asyncio.Semaphore binds to whichever loop first awaits it;
sharing one instance across loops in different threads silently deadlocks
because cross-loop wake-ups aren't thread-safe. Capture the running loop on
first acquire and raise a clear RuntimeError on mismatch — preferable to
the silent hang the audit identified.

The mismatch check uses double-check-with-lock to make the cold-path first
capture race-free across threads. Same-loop fast path is lock-free.

Module docstring grows a paragraph mirroring the sync Bulkhead's per-world
warning; class docstring cross-links it. Three tests added:
  - first_acquire_captures_running_loop (capture happens)
  - same_loop_succeeds_across_multiple_acquires (no false positive)
  - cross_loop_acquire_raises_runtimeerror (two asyncio.run() calls deterministically)

Closes Medium #M3 + #M4 in planning/audit/2026-06-07-deep-audit.md (bundled
per the audit's "keep this fix and the docstring fix in the same PR" note).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: lesnik512

src/httpware/middleware/resilience/bulkhead.py

@@ -9,6 +9,13 @@
 
 AsyncBulkhead is the sharable unit — pass the same instance to multiple
 AsyncClient(middleware=[shared]) calls to enforce a joint cap across clients.
+
+AsyncBulkhead is single-event-loop: the underlying asyncio.Semaphore binds
+to whichever loop first awaits it, and cross-loop wake-ups are not thread
+safe. A single instance acquired from a second event loop (e.g. another
+thread running asyncio.run) raises RuntimeError on entry rather than
+deadlocking silently. To cap a sync+async or cross-thread workload, use
+a Bulkhead and an AsyncBulkhead with matching max_concurrent.
 """
 
 import asyncio
@@ -24,6 +31,11 @@
 
 _MAX_CONCURRENT_INVALID = "max_concurrent must be >= 1"
 _ACQUIRE_TIMEOUT_INVALID = "acquire_timeout must be >= 0"
+_ASYNCBULKHEAD_CROSS_LOOP_MSG = (
+    "AsyncBulkhead is bound to a single event loop. First seen on {first!r}; "
+    "current request is on {current!r}. Use one AsyncBulkhead per loop; "
+    "cross-thread sharing requires the sync Bulkhead primitive."
+)
 
 _LOGGER = logging.getLogger("httpware.bulkhead")
 
@@ -42,7 +54,8 @@ class AsyncBulkhead:
         Defaults to ``1.0``. ``None`` waits forever; ``0`` fails fast. Must be
         ``>= 0`` (or ``None``).
 
-    See the module docstring for the algorithm and middleware-ordering guidance.
+    See the module docstring for the algorithm, middleware-ordering guidance,
+    and the single-event-loop constraint.
 
     """
 
@@ -59,9 +72,29 @@ def __init__(
         self._max_concurrent = max_concurrent
         self._acquire_timeout = acquire_timeout
         self._sem = asyncio.Semaphore(max_concurrent)
+        self._loop: asyncio.AbstractEventLoop | None = None
+        self._loop_lock = threading.Lock()
+
+    def _check_loop(self) -> None:
+        current = asyncio.get_running_loop()
+        cached = self._loop
+        if cached is current:
+            return
+        if cached is not None:
+            raise RuntimeError(
+                _ASYNCBULKHEAD_CROSS_LOOP_MSG.format(first=cached, current=current),
+            )
+        with self._loop_lock:
+            if self._loop is None:
+                self._loop = current
+            elif self._loop is not current:
+                raise RuntimeError(
+                    _ASYNCBULKHEAD_CROSS_LOOP_MSG.format(first=self._loop, current=current),
+                )
 
     async def __call__(self, request: httpx2.Request, next: AsyncNext) -> httpx2.Response:  # noqa: A002
         """Acquire a slot (bounded by acquire_timeout), invoke next, release."""
+        self._check_loop()
         try:
             if self._acquire_timeout is None:
                 await self._sem.acquire()

tests/test_bulkhead.py

@@ -431,3 +431,45 @@ async def slow_handler(request: httpx2.Request) -> httpx2.Response:
     assert record.acquire_timeout == 0.0  # ty: ignore[unresolved-attribute]
     assert record.method == "GET"  # ty: ignore[unresolved-attribute]
     assert "example.test/y" in record.url  # ty: ignore[unresolved-attribute]
+
+
+# ───── Single-event-loop guard ──────────────────────────────────────────────
+
+
+async def test_first_acquire_captures_running_loop() -> None:
+    """AsyncBulkhead binds to whichever loop first acquires a slot."""
+    bulkhead = AsyncBulkhead(max_concurrent=_MAX_CONCURRENT_1)
+    assert bulkhead._loop is None  # noqa: SLF001
+    handler = _SlowHandler(delay=0.0)
+    async with _client(handler, bulkhead=bulkhead) as client:
+        await client.get("https://example.test/x")
+    assert bulkhead._loop is asyncio.get_running_loop()  # noqa: SLF001
+
+
+async def test_same_loop_succeeds_across_multiple_acquires() -> None:
+    """Repeated acquires on the same loop never trigger the cross-loop guard."""
+    bulkhead = AsyncBulkhead(max_concurrent=_MAX_CONCURRENT_2)
+    handler = _SlowHandler(delay=0.0)
+    async with _client(handler, bulkhead=bulkhead) as client:
+        for _ in range(5):
+            response = await client.get("https://example.test/x")
+            assert response.status_code == HTTPStatus.OK
+
+
+def test_cross_loop_acquire_raises_runtimeerror() -> None:
+    """A bulkhead first used on one loop, then reused on another, raises RuntimeError.
+
+    Each asyncio.run() call creates a fresh event loop and tears it down on
+    exit. Sharing one AsyncBulkhead instance across two asyncio.run() calls
+    is the cross-loop case the guard prevents.
+    """
+    bulkhead = AsyncBulkhead(max_concurrent=_MAX_CONCURRENT_1)
+    handler = _SlowHandler(delay=0.0)
+
+    async def _run_once() -> None:
+        async with _client(handler, bulkhead=bulkhead) as client:
+            await client.get("https://example.test/x")
+
+    asyncio.run(_run_once())  # captures loop L1, then L1 closes
+    with pytest.raises(RuntimeError, match="AsyncBulkhead is bound to a single event loop"):
+        asyncio.run(_run_once())