@@ -0,0 +1,166 @@
+---
+summary: Share Bulkhead/AsyncBulkhead's hand-duplicated constructor validation and rejection-event-plus-BulkheadFullError construction via two module-level functions.
+---
+
+# Design: Share Bulkhead's validation and rejection logic
+
+## Summary
+
+`AsyncBulkhead` and `Bulkhead` (`src/httpware/middleware/resilience/bulkhead.py`,
+185 lines) hand-duplicate two blocks: constructor validation
+(`max_concurrent`/`acquire_timeout` range checks) and the on-timeout
+rejection path (emit a `bulkhead.rejected` event, construct
+`BulkheadFullError`). Only the semaphore type and acquire-timeout
+mechanics genuinely differ. This change extracts both blocks into two
+module-level functions in the same file — `_validate_bulkhead_config` and
+`_emit_bulkhead_rejected` — closing a deferral this repo's own planning
+history named for itself. No behavior change.
+
+## Motivation
+
+- `planning/changes/2026-06-23.01-retry-policy-extraction.md`'s
+  Out-of-scope section states: "Not extending the same treatment to
+  Bulkhead in this change." That change gave `circuit_breaker.py` and
+  `retry.py` shared decision objects (`_CircuitBreakerState`,
+  `_RetryPolicy`); `bulkhead.py` never got the equivalent treatment.
+- **Verified duplication:** `AsyncBulkhead.__init__` (`:62-76`) and
+  `Bulkhead.__init__` (`:143-155`) run the identical two `if`/`raise
+  ValueError` checks. `AsyncBulkhead.__call__`'s `except TimeoutError`
+  branch (`:107-123`) and `Bulkhead.__call__`'s `if not acquired:` branch
+  (`:163-179`) build the identical `_emit_event(...)` call (same event
+  name, level, message, attributes shape) and the identical
+  `BulkheadFullError(max_concurrent=..., acquire_timeout=...)`
+  construction.
+- **Deletion test:** delete the two shared functions and both blocks
+  reappear verbatim at all four call sites (2 `__init__`s, 2 `__call__`s)
+  — real, load-bearing logic, not a pass-through.
+
+## Non-goals
+
+- No behavior change. Validation error messages, the emitted event's
+  name/level/message/attributes, and `BulkheadFullError`'s fields stay
+  byte-identical.
+- Not a stateful class (unlike `_CircuitBreakerState`/`_RetryPolicy`).
+  Both extracted operations are one-shot/per-call pure functions with no
+  config or evolving state reused across calls — a class would be a
+  namespace with no benefit, the same reasoning already applied to
+  `client.py`'s request-assembly extraction earlier this session.
+- Not moving the functions to `_internal/`. `bulkhead.py`'s logic is used
+  only by its own two classes — same situation as `_CircuitBreakerState`
+  in `circuit_breaker.py` and `_RetryPolicy` in `retry.py`, both of which
+  stayed local to their file.
+- Not touching `AsyncBulkhead._check_loop`, the semaphore acquire
+  mechanics, or anything else in either `__call__`/`__init__` beyond the
+  two blocks named above.
+
+## Design
+
+### 1. `_validate_bulkhead_config`
+
+```python
+def _validate_bulkhead_config(*, max_concurrent: int, acquire_timeout: float | None) -> None:
+    if max_concurrent < 1:
+        raise ValueError(_MAX_CONCURRENT_INVALID)
+    if acquire_timeout is not None and acquire_timeout < 0:
+        raise ValueError(_ACQUIRE_TIMEOUT_INVALID)
+```
+
+Replaces the two `if`/`raise` lines in both `__init__`s with:
+
+```python
+_validate_bulkhead_config(max_concurrent=max_concurrent, acquire_timeout=acquire_timeout)
+```
+
+Uses `-> None` (not `typing.NoReturn`), matching the existing house idiom
+for validation helpers that raise conditionally
+(`_validate_max_response_body_bytes`, `_validate_httpx2_client_conflict`)
+— no precedent for `NoReturn` exists in this codebase, and this function
+doesn't always raise (only on the invalid branches), so `None` is also the
+technically correct annotation here regardless of style.
+
+### 2. `_emit_bulkhead_rejected` — builds and returns, does not raise
+
+```python
+def _emit_bulkhead_rejected(
+    request: httpx2.Request,
+    *,
+    max_concurrent: int,
+    acquire_timeout: float | None,
+) -> BulkheadFullError:
+    _emit_event(
+        _LOGGER,
+        "bulkhead.rejected",
+        level=logging.WARNING,
+        message="bulkhead rejected request — acquire_timeout exceeded",
+        attributes={
+            "max_concurrent": max_concurrent,
+            "acquire_timeout": acquire_timeout,
+            "method": request.method,
+            "url": str(request.url),
+        },
+    )
+    return BulkheadFullError(max_concurrent=max_concurrent, acquire_timeout=acquire_timeout)
+```
+
+**Deliberately returns the exception rather than raising it.** The two
+call sites need different exception-chaining behavior: `AsyncBulkhead`'s
+rejection fires inside `except TimeoutError as exc:` and today does
+`raise BulkheadFullError(...) from exc` (explicit `__cause__`);
+`Bulkhead`'s rejection fires inside `if not acquired:` — no active
+exception, bare `raise BulkheadFullError(...)`. If the function raised
+internally, the async call site's explicit `from exc` would be lost,
+silently downgrading to implicit `__context__` chaining (a different
+traceback presentation — "the above exception was the direct cause of"
+vs "during handling of the above exception"). Returning the constructed
+exception lets each call site keep its own `raise` statement exactly as
+today:
+
+```python
+# AsyncBulkhead.__call__, inside except TimeoutError as exc:
+raise _emit_bulkhead_rejected(
+    request, max_concurrent=self._max_concurrent, acquire_timeout=self._acquire_timeout,
+) from exc
+
+# Bulkhead.__call__, inside if not acquired:
+raise _emit_bulkhead_rejected(
+    request, max_concurrent=self._max_concurrent, acquire_timeout=self._acquire_timeout,
+)
+```
+
+### 3. Placement
+
+Both functions go in `bulkhead.py`, module level, alongside the existing
+module constants (`_MAX_CONCURRENT_INVALID`, `_ACQUIRE_TIMEOUT_INVALID`,
+`_LOGGER`) — before `class AsyncBulkhead:`.
+
+## Testing
+
+- **Parity net:** existing `tests/test_bulkhead.py` and
+  `tests/test_bulkhead_sync.py` stay green with the assertions they
+  already have (message content, `BulkheadFullError` field values,
+  `bulkhead.rejected` event via `caplog`) — byte-identical behavior is the
+  bar.
+- **New chaining tests** (the one subtlety this refactor is careful to
+  preserve, and no existing test asserts it): in `test_bulkhead.py`, add
+  an assertion that the raised `BulkheadFullError`'s `__cause__` is the
+  `TimeoutError` that triggered it; in `test_bulkhead_sync.py`, add an
+  assertion that `__cause__` is `None`. Mirrors how `DecodeError.__cause__`
+  is explicitly tested in `test_errors.py`.
+- **No new direct-unit-test file** for the two shared functions — the
+  existing construction-error tests already exercise
+  `_validate_bulkhead_config` fully (both branches, both classes), and
+  the existing `BulkheadFullError`-field/event tests plus the new
+  chaining tests already exercise `_emit_bulkhead_rejected` fully.
+- `just lint && just test` both clean; 100% coverage maintained.
+
+## Risk
+
+- **Chaining regression** (the main risk this design addresses directly):
+  *Mitigation:* returning rather than raising keeps each `raise` statement
+  exactly where it is today; the new chaining tests make a future
+  regression (e.g. someone "simplifying" the sync call site to also use
+  `from exc` where no exception exists, or vice versa) fail loudly.
+- **Behavioral drift in the shared validation/event logic** (unlikely ×
+  low): *Mitigation:* extract verbatim under the existing green suites,
+  which assert exact message text and field values; do not edit those
+  assertions in this change (only add the two new chaining assertions).
@@ -40,6 +40,34 @@
 _LOGGER = logging.getLogger("httpware.bulkhead")
 
 
+def _validate_bulkhead_config(*, max_concurrent: int, acquire_timeout: float | None) -> None:
+    if max_concurrent < 1:
+        raise ValueError(_MAX_CONCURRENT_INVALID)
+    if acquire_timeout is not None and acquire_timeout < 0:
+        raise ValueError(_ACQUIRE_TIMEOUT_INVALID)
+
+
+def _emit_bulkhead_rejected(
+    request: httpx2.Request,
+    *,
+    max_concurrent: int,
+    acquire_timeout: float | None,
+) -> BulkheadFullError:
+    _emit_event(
+        _LOGGER,
+        "bulkhead.rejected",
+        level=logging.WARNING,
+        message="bulkhead rejected request — acquire_timeout exceeded",
+        attributes={
+            "max_concurrent": max_concurrent,
+            "acquire_timeout": acquire_timeout,
+            "method": request.method,
+            "url": str(request.url),
+        },
+    )
+    return BulkheadFullError(max_concurrent=max_concurrent, acquire_timeout=acquire_timeout)
+
+
 class AsyncBulkhead:
     """Async concurrency limiter middleware backed by ``asyncio.Semaphore``.
 
@@ -65,10 +93,7 @@ def __init__(
         max_concurrent: int,
         acquire_timeout: float | None = 1.0,
     ) -> None:
-        if max_concurrent < 1:
-            raise ValueError(_MAX_CONCURRENT_INVALID)
-        if acquire_timeout is not None and acquire_timeout < 0:
-            raise ValueError(_ACQUIRE_TIMEOUT_INVALID)
+        _validate_bulkhead_config(max_concurrent=max_concurrent, acquire_timeout=acquire_timeout)
         self._max_concurrent = max_concurrent
         self._acquire_timeout = acquire_timeout
         self._sem = asyncio.Semaphore(max_concurrent)
@@ -105,19 +130,8 @@ async def __call__(self, request: httpx2.Request, next: AsyncNext) -> httpx2.Res
                 async with asyncio.timeout(self._acquire_timeout):
                     await self._sem.acquire()
         except TimeoutError as exc:
-            _emit_event(
-                _LOGGER,
-                "bulkhead.rejected",
-                level=logging.WARNING,
-                message="bulkhead rejected request — acquire_timeout exceeded",
-                attributes={
-                    "max_concurrent": self._max_concurrent,
-                    "acquire_timeout": self._acquire_timeout,
-                    "method": request.method,
-                    "url": str(request.url),
-                },
-            )
-            raise BulkheadFullError(
+            raise _emit_bulkhead_rejected(
+                request,
                 max_concurrent=self._max_concurrent,
                 acquire_timeout=self._acquire_timeout,
             ) from exc
@@ -146,10 +160,7 @@ def __init__(
         max_concurrent: int,
         acquire_timeout: float | None = 1.0,
     ) -> None:
-        if max_concurrent < 1:
-            raise ValueError(_MAX_CONCURRENT_INVALID)
-        if acquire_timeout is not None and acquire_timeout < 0:
-            raise ValueError(_ACQUIRE_TIMEOUT_INVALID)
+        _validate_bulkhead_config(max_concurrent=max_concurrent, acquire_timeout=acquire_timeout)
         self._max_concurrent = max_concurrent
         self._acquire_timeout = acquire_timeout
         self._sem = threading.Semaphore(max_concurrent)
@@ -161,19 +172,8 @@ def __call__(self, request: httpx2.Request, next: Next) -> httpx2.Response:  # n
         # False otherwise). Both match AsyncBulkhead's contract.
         acquired = self._sem.acquire(timeout=self._acquire_timeout)
         if not acquired:
-            _emit_event(
-                _LOGGER,
-                "bulkhead.rejected",
-                level=logging.WARNING,
-                message="bulkhead rejected request — acquire_timeout exceeded",
-                attributes={
-                    "max_concurrent": self._max_concurrent,
-                    "acquire_timeout": self._acquire_timeout,
-                    "method": request.method,
-                    "url": str(request.url),
-                },
-            )
-            raise BulkheadFullError(
+            raise _emit_bulkhead_rejected(
+                request,
                 max_concurrent=self._max_concurrent,
                 acquire_timeout=self._acquire_timeout,
             )
 
@@ -148,6 +148,30 @@ async def _hold_slot() -> None:
         await task
 
 
+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)
+    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)
+
+    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
+
+
 async def test_bounded_wait_raises_bulkhead_full_error() -> None:
     """With max_concurrent=1 and acquire_timeout=0.02, the second call raises after ~20ms.
 
 
@@ -129,6 +129,26 @@ def test_acquire_timeout_rejects_when_no_slot_available() -> None:
         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)
+    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()
+    # Give the holder time to acquire the only slot
+    time.sleep(0.01)
+    try:
+        with pytest.raises(BulkheadFullError) as exc_info:
+            client.get("https://example.test/blocked")
+        assert exc_info.value.__cause__ is None
+    finally:
+        holder.join()
+
+
 def test_releases_slot_on_exception() -> None:
     """A handler that raises must still cause the slot to be released."""
     calls = []