diff --git a/.gitignore b/.gitignore
index ea529a2..6b649cf 100644
--- a/.gitignore
+++ b/.gitignore
@@ -11,6 +11,7 @@ __pycache__/*
 .env
 .pytest_cache
 .ruff_cache
+.hypothesis/
 .coverage
 htmlcov/
 coverage.xml
diff --git a/planning/deferred-work.md b/planning/deferred-work.md
index 9715f3c..240ac79 100644
--- a/planning/deferred-work.md
+++ b/planning/deferred-work.md
@@ -4,6 +4,10 @@ Items raised in reviews that are real but not actionable now.
 
 ## Open
 
+### Retry + streaming bodies (Epic 4 interaction)
+
+- **`Retry` re-invokes `next(request)` with the same `httpx2.Request` on each attempt.** Safe for in-memory bytes/JSON bodies; unsafe for streaming/async-iterable bodies (consumed iterator can't replay). When Epic 4 ships `AsyncClient.stream` (`4-3`), Retry needs to refuse to retry streamed-body requests (or document that callers supply a body factory). Spec: `planning/specs/2026-06-05-retry-and-retry-budget-design.md` §"Open questions".
+
 ### Decoder-side
 
 - **`_get_adapter` `lru_cache` is module-global, not per-decoder instance** — keyed by `model` only; two `PydanticDecoder()` instances with different configurations (none today) would share adapters, and the cache survives across tests unless explicitly cleared. Revisit if/when a configurable `PydanticDecoder(mode=..., strict=...)` lands. (`src/httpware/decoders/pydantic.py:12-14`)
diff --git a/planning/engineering.md b/planning/engineering.md
index f94a483..1a06e56 100644
--- a/planning/engineering.md
+++ b/planning/engineering.md
@@ -121,7 +121,9 @@ Post-pivot, the roadmap has three categories. Topic slugs in `planning/specs/` a
 
 ### Surviving (land in subsequent PRs)
 
-- **Epic 3 — Resilience:** `3-1` per-attempt timeout, `3-2` retry, `3-3` `RetryBudget`, `3-4` `RetryBudget` middleware integration, `3-5` `Bulkhead`, `3-6` extension-slot docs.
+- **Epic 3 — Resilience:**
+  - **Shipped in v0.4 slice 1:** `Retry` middleware + Finagle-style `RetryBudget` token bucket + `attempt_timeout=` parameter (folded-in 3-1). See [`planning/specs/2026-06-05-retry-and-retry-budget-design.md`](specs/2026-06-05-retry-and-retry-budget-design.md) and [`planning/plans/2026-06-05-retry-and-retry-budget-plan.md`](plans/2026-06-05-retry-and-retry-budget-plan.md).
+  - **Remaining:** `3-5` `Bulkhead`, `3-6` extension-slot docs.
 - **Epic 4 — Streaming:** `4-3` `AsyncClient.stream` context manager (forwards to `httpx2.AsyncClient.stream`; no `StreamResponse` type).
 - **Epic 5 — Observability:** `5-1` Layer 1 middleware hooks, `5-2` wire into resilience middlewares, `5-4` OpenTelemetry middleware (`otel` extra), `5-5` logging policy CI grep.
 - **Epic 6 — Ship v1.0:** `6-2` docs site (`mkdocs`), `6-3` benchmarks, `6-5` release flow (Trusted Publishers + Sigstore).
diff --git a/planning/plans/2026-06-05-retry-and-retry-budget-plan.md b/planning/plans/2026-06-05-retry-and-retry-budget-plan.md
index ba4c7f0..120c3f8 100644
--- a/planning/plans/2026-06-05-retry-and-retry-budget-plan.md
+++ b/planning/plans/2026-06-05-retry-and-retry-budget-plan.md
@@ -63,12 +63,14 @@ mkdir -p src/httpware/middleware/resilience
 
 Then create each file with the contents below. Use the Write tool, not bash heredocs.
 
-`src/httpware/middleware/resilience/__init__.py`:
+`src/httpware/middleware/resilience/__init__.py` (docstring-only — re-exports defer to Task 7 so intermediate tasks can `import httpware.middleware.resilience.budget` without tripping an import-time `ImportError` from this `__init__.py`):
 ```python
-"""Resilience primitives: Retry middleware and RetryBudget token bucket."""
+"""Resilience primitives: Retry middleware and RetryBudget token bucket.
 
-from httpware.middleware.resilience.budget import RetryBudget
-from httpware.middleware.resilience.retry import Retry
+Re-exports land in Task 7 once both classes exist; until then this file is
+docstring-only so that importing ``httpware.middleware.resilience.budget``
+during the intermediate tasks does not trip an import-time ``ImportError``.
+"""
 ```
 
 `src/httpware/middleware/resilience/budget.py`:
@@ -137,7 +139,11 @@ Expected: FAIL (`ImportError: cannot import name 'NetworkError'`).
 Edit `src/httpware/errors.py`. Add a new class immediately after the existing `class TransportError`:
 ```python
 class NetworkError(TransportError):
-    """Transient network-layer failure (connect/read/write/pool). Safe to retry."""
+    """Transient network-layer failure (connect/read/write/close). Safe to retry.
+
+Pool-acquisition timeouts are NOT under this class; they raise ``TimeoutError``
+via ``httpx2.PoolTimeout`` (a ``TimeoutException`` subclass).
+"""
 ```
 
 Run: `uv run pytest tests/test_errors.py::test_network_error_is_transport_error -v`
@@ -218,7 +224,7 @@ Becomes:
 
 The `httpx2.NetworkError` branch must come BEFORE `httpx2.HTTPError` (HTTPError is the broader base). `httpx2.NetworkError` is httpx's documented base for `ConnectError`, `ReadError`, `WriteError`, `PoolTimeout` — if `httpx2`'s symbol name differs (e.g., `httpx2.exceptions.NetworkError`), use whichever import path mirrors the existing `httpx2.ConnectError` import in `tests/test_error_mapping_terminal.py` (which works via top-level `httpx2`).
 
-If `httpx2.NetworkError` does not exist, fall back to enumerating the transient subset explicitly: `except (httpx2.ConnectError, httpx2.ReadError, httpx2.WriteError, httpx2.PoolTimeout) as exc:`. The plan author has confirmed `httpx2.ConnectError` and `httpx2.ReadTimeout` already work in the existing tests; the enumeration fallback is safe.
+If `httpx2.NetworkError` does not exist, fall back to enumerating the transient subset explicitly: `except (httpx2.ConnectError, httpx2.ReadError, httpx2.WriteError, httpx2.CloseError) as exc:`. (`PoolTimeout` is NOT in this list — it inherits from `httpx2.TimeoutException` and is already caught by the timeout branch above.) The plan author has confirmed `httpx2.ConnectError` and `httpx2.ReadTimeout` already work in the existing tests; the enumeration fallback is safe.
 
 - [ ] **Step 6: Run the new terminal-mapping test**
 
@@ -265,7 +271,7 @@ git add src/httpware/errors.py src/httpware/client.py tests/test_errors.py tests
 git commit -m "feat(errors): add NetworkError(TransportError) for transient httpx2 failures
 
 Refines _terminal so httpx2.NetworkError-family exceptions (ConnectError, ReadError,
-WriteError, PoolTimeout) map to httpware.NetworkError. InvalidURL and CookieConflict
+WriteError, CloseError) map to httpware.NetworkError. InvalidURL and CookieConflict
 stay bare TransportError. Prerequisite for the Retry middleware so it can retry
 transient failures without retrying typos."
 ```
@@ -1029,15 +1035,30 @@ class Retry:
 Run: `uv run pytest tests/test_retry.py -v`
 Expected: all PASS.
 
-- [ ] **Step 4: Lint**
+- [ ] **Step 4: Wire `Retry` + `RetryBudget` into `resilience/__init__.py`**
 
-Run: `uv run ruff check src/httpware/middleware/resilience/retry.py tests/test_retry.py && uv run ty check src/httpware/middleware/resilience/retry.py`
+Now that both classes exist, replace `src/httpware/middleware/resilience/__init__.py` with:
+```python
+"""Resilience primitives: Retry middleware and RetryBudget token bucket."""
+
+from httpware.middleware.resilience.budget import RetryBudget
+from httpware.middleware.resilience.retry import Retry
+
+
+__all__ = ["Retry", "RetryBudget"]
+```
+
+The `__all__` is required to silence ruff F401 ("imported but unused") and matches the pattern used by `httpware/__init__.py` and `httpware/decoders/__init__.py`.
+
+- [ ] **Step 5: Lint**
+
+Run: `uv run ruff check src/httpware/middleware/resilience/ tests/test_retry.py && uv run ty check src/httpware/middleware/resilience/`
 Expected: clean. If ruff flags `Callable` / `Awaitable` import paths, adjust per existing project pattern (see `middleware/__init__.py` which uses `from collections.abc import Awaitable, Callable`).
 
-- [ ] **Step 5: Stage and commit**
+- [ ] **Step 6: Stage and commit**
 
 ```bash
-git add src/httpware/middleware/resilience/retry.py tests/test_retry.py
+git add src/httpware/middleware/resilience/retry.py src/httpware/middleware/resilience/__init__.py tests/test_retry.py
 git commit -m "feat(resilience): Retry middleware — status-code retry + exhaustion
 
 Covers: happy path, 503-then-200, max_attempts exhaustion with PEP-678 note,
diff --git a/planning/specs/2026-06-05-retry-and-retry-budget-design.md b/planning/specs/2026-06-05-retry-and-retry-budget-design.md
index 1aba78d..05fdf9a 100644
--- a/planning/specs/2026-06-05-retry-and-retry-budget-design.md
+++ b/planning/specs/2026-06-05-retry-and-retry-budget-design.md
@@ -28,7 +28,7 @@ The current `AsyncClient._terminal` maps every non-timeout `httpx2.HTTPError` (i
 ```python
 # In src/httpware/errors.py:
 class NetworkError(TransportError):
-    """Transient network-layer failure (connect / read / write / pool). Safe to retry."""
+    """Transient network-layer failure (connect / read / write / close). Safe to retry."""
 ```
 
 And refines the terminal mapping so that `httpx2`'s transient-network exception family (`httpx2.NetworkError` per httpx convention, or whichever symbols httpx2 exposes for the same hierarchy) raises `httpware.NetworkError` rather than the broader `TransportError`. `InvalidURL` and `CookieConflict` continue to raise `TransportError` directly so they are NOT retried. Existing tests catching `TransportError` keep working (`NetworkError` is a subclass).
diff --git a/src/httpware/__init__.py b/src/httpware/__init__.py
index 5d8e9ba..b80c90d 100644
--- a/src/httpware/__init__.py
+++ b/src/httpware/__init__.py
@@ -10,8 +10,10 @@
     ConflictError,
     ForbiddenError,
     InternalServerError,
+    NetworkError,
     NotFoundError,
     RateLimitedError,
+    RetryBudgetExhaustedError,
     ServerStatusError,
     ServiceUnavailableError,
     StatusError,
@@ -21,6 +23,7 @@
     UnprocessableEntityError,
 )
 from httpware.middleware import Middleware, Next, after_response, before_request, on_error
+from httpware.middleware.resilience import Retry, RetryBudget
 
 
 __all__ = [
@@ -33,10 +36,14 @@
     "ForbiddenError",
     "InternalServerError",
     "Middleware",
+    "NetworkError",
     "Next",
     "NotFoundError",
     "RateLimitedError",
     "ResponseDecoder",
+    "Retry",
+    "RetryBudget",
+    "RetryBudgetExhaustedError",
     "ServerStatusError",
     "ServiceUnavailableError",
     "StatusError",
diff --git a/src/httpware/client.py b/src/httpware/client.py
index c92c2df..1a355de 100644
--- a/src/httpware/client.py
+++ b/src/httpware/client.py
@@ -11,6 +11,7 @@
 from httpware.errors import (
     STATUS_TO_EXCEPTION,
     ClientStatusError,
+    NetworkError,
     ServerStatusError,
     TimeoutError,  # noqa: A004
     TransportError,
@@ -110,6 +111,8 @@ async def _terminal(self, request: httpx2.Request) -> httpx2.Response:
             raise TimeoutError(str(exc)) from exc
         except (httpx2.InvalidURL, httpx2.CookieConflict) as exc:
             raise TransportError(str(exc)) from exc
+        except httpx2.NetworkError as exc:
+            raise NetworkError(str(exc)) from exc
         except httpx2.HTTPError as exc:
             raise TransportError(str(exc)) from exc
         except RuntimeError as exc:
diff --git a/src/httpware/errors.py b/src/httpware/errors.py
index 321630d..1c8cbcc 100644
--- a/src/httpware/errors.py
+++ b/src/httpware/errors.py
@@ -40,6 +40,14 @@ class TransportError(ClientError):
     """Connection / network / protocol failure raised before a response was received."""
 
 
+class NetworkError(TransportError):
+    """Transient network-layer failure (connect/read/write/close). Safe to retry.
+
+    Pool-acquisition timeouts are NOT under this class; they raise ``TimeoutError``
+    via ``httpx2.PoolTimeout`` (a ``TimeoutException`` subclass).
+    """
+
+
 class TimeoutError(ClientError, builtins.TimeoutError):  # noqa: A001
     """Client-side timeout (connect / read / write / pool).
 
@@ -136,3 +144,42 @@ class ServiceUnavailableError(ServerStatusError):
     500: InternalServerError,
     503: ServiceUnavailableError,
 }
+
+
+def _reconstruct_budget_exhausted(
+    cls: "type[RetryBudgetExhaustedError]",
+    last_response: httpx2.Response | None,
+    last_exception: BaseException | None,
+    attempts: int,
+) -> "RetryBudgetExhaustedError":
+    return cls(last_response=last_response, last_exception=last_exception, attempts=attempts)
+
+
+class RetryBudgetExhaustedError(ClientError):
+    """Raised when a retry was needed but the RetryBudget refused to permit it.
+
+    Carries the last response and/or exception observed before the budget refused,
+    plus the number of attempts already completed.
+    """
+
+    last_response: httpx2.Response | None
+    last_exception: BaseException | None
+    attempts: int
+
+    def __init__(
+        self,
+        *,
+        last_response: httpx2.Response | None,
+        last_exception: BaseException | None,
+        attempts: int,
+    ) -> None:
+        self.last_response = last_response
+        self.last_exception = last_exception
+        self.attempts = attempts
+        super().__init__(f"retry budget exhausted after {attempts} attempt(s)")
+
+    def __reduce__(self) -> tuple[Any, ...]:
+        return (
+            _reconstruct_budget_exhausted,
+            (type(self), self.last_response, self.last_exception, self.attempts),
+        )
diff --git a/src/httpware/middleware/resilience/__init__.py b/src/httpware/middleware/resilience/__init__.py
new file mode 100644
index 0000000..61244be
--- /dev/null
+++ b/src/httpware/middleware/resilience/__init__.py
@@ -0,0 +1,7 @@
+"""Resilience primitives: Retry middleware and RetryBudget token bucket."""
+
+from httpware.middleware.resilience.budget import RetryBudget
+from httpware.middleware.resilience.retry import Retry
+
+
+__all__ = ["Retry", "RetryBudget"]
diff --git a/src/httpware/middleware/resilience/_backoff.py b/src/httpware/middleware/resilience/_backoff.py
new file mode 100644
index 0000000..f9050ca
--- /dev/null
+++ b/src/httpware/middleware/resilience/_backoff.py
@@ -0,0 +1,26 @@
+"""Full-jitter exponential backoff helper (private)."""
+
+import random
+from collections.abc import Callable
+
+
+def full_jitter_delay(
+    attempt_index: int,
+    *,
+    base_delay: float,
+    max_delay: float,
+    _random_uniform: Callable[[float, float], float] = random.uniform,
+) -> float:
+    """Return a backoff delay using AWS's "full jitter" formulation.
+
+    sleep = uniform(0, min(max_delay, base_delay * 2.0 ** attempt_index))
+
+    `attempt_index` is 0 for the first retry, 1 for the second, etc.
+
+    Uses ``2.0 **`` (float exponentiation) rather than ``2 **`` so that
+    ``attempt_index >= 1024`` saturates to ``math.inf`` and ``min`` clamps to
+    ``max_delay`` — ``2 ** 1024`` would raise ``OverflowError`` during the
+    int→float conversion.
+    """
+    ceiling = min(max_delay, base_delay * (2.0**attempt_index))
+    return _random_uniform(0.0, ceiling)
diff --git a/src/httpware/middleware/resilience/budget.py b/src/httpware/middleware/resilience/budget.py
new file mode 100644
index 0000000..16c8be9
--- /dev/null
+++ b/src/httpware/middleware/resilience/budget.py
@@ -0,0 +1,64 @@
+"""Finagle-style token-bucket retry budget.
+
+See planning/specs/2026-06-05-retry-and-retry-budget-design.md for the contract.
+No locking: asyncio runs coroutines cooperatively on a single thread, so deque
+mutations between await points are atomic with respect to other coroutines on
+the same event loop. Cross-thread use is out of scope.
+"""
+
+import time
+from collections import deque
+from collections.abc import Callable
+
+
+class RetryBudget:
+    """Token-bucket budget bounding retry rate to prevent retry storms.
+
+    Each request deposits a token; each retry attempts to withdraw one.
+    Available retries are bounded by `percent_can_retry` of recent deposits,
+    plus a `min_retries_per_sec * ttl` floor.
+    """
+
+    def __init__(
+        self,
+        *,
+        ttl: float = 10.0,
+        min_retries_per_sec: float = 10.0,
+        percent_can_retry: float = 0.2,
+        _now: Callable[[], float] = time.monotonic,
+    ) -> None:
+        self._ttl = ttl
+        self._min_retries_per_sec = min_retries_per_sec
+        self._percent_can_retry = percent_can_retry
+        self._now = _now
+        self._deposits: deque[float] = deque()
+        self._withdrawn: deque[float] = deque()
+
+    def _purge(self, now: float) -> None:
+        # Strict `< cutoff` keeps entries at exactly `now - ttl`: window is [now - ttl, now].
+        cutoff = now - self._ttl
+        while self._deposits and self._deposits[0] < cutoff:
+            self._deposits.popleft()
+        while self._withdrawn and self._withdrawn[0] < cutoff:
+            self._withdrawn.popleft()
+
+    def deposit(self) -> None:
+        """Record a request (success or failure attempt). Adds one token."""
+        now = self._now()
+        self._purge(now)
+        self._deposits.append(now)
+
+    def try_withdraw(self) -> bool:
+        """Atomically attempt to spend one retry token.
+
+        Returns True if a retry is permitted, False if the budget is exhausted.
+        Never blocks.
+        """
+        now = self._now()
+        self._purge(now)
+        floor = int(self._min_retries_per_sec * self._ttl)
+        ceiling = int(len(self._deposits) * self._percent_can_retry) + floor
+        if len(self._withdrawn) >= ceiling:
+            return False
+        self._withdrawn.append(now)
+        return True
diff --git a/src/httpware/middleware/resilience/retry.py b/src/httpware/middleware/resilience/retry.py
new file mode 100644
index 0000000..43ef1a2
--- /dev/null
+++ b/src/httpware/middleware/resilience/retry.py
@@ -0,0 +1,158 @@
+"""Retry middleware — automatic retry of transient failures with budget control.
+
+See planning/specs/2026-06-05-retry-and-retry-budget-design.md for the full contract.
+
+Status-code retry: the AsyncClient terminal raises StatusError subclasses on 4xx/5xx,
+so Retry catches StatusError and inspects exc.response.status_code. The original
+StatusError subclass is re-raised unwrapped on exhaustion, with a PEP 678 note added.
+"""
+
+import asyncio
+import builtins
+import datetime
+import email.utils
+from collections.abc import Awaitable, Callable
+from http import HTTPStatus
+
+import httpx2
+
+from httpware.errors import NetworkError, RetryBudgetExhaustedError, StatusError, TimeoutError  # noqa: A004
+from httpware.middleware import Next
+from httpware.middleware.resilience._backoff import full_jitter_delay
+from httpware.middleware.resilience.budget import RetryBudget
+
+
+DEFAULT_RETRY_STATUS_CODES = frozenset(
+    {
+        int(HTTPStatus.REQUEST_TIMEOUT),
+        int(HTTPStatus.TOO_MANY_REQUESTS),
+        int(HTTPStatus.BAD_GATEWAY),
+        int(HTTPStatus.SERVICE_UNAVAILABLE),
+        int(HTTPStatus.GATEWAY_TIMEOUT),
+    }
+)
+
+DEFAULT_IDEMPOTENT_METHODS = frozenset(
+    {
+        "GET",
+        "HEAD",
+        "OPTIONS",
+        "PUT",
+        "DELETE",
+    }
+)
+
+_MAX_ATTEMPTS_INVALID = "max_attempts must be >= 1"
+
+
+def _parse_retry_after(value: str) -> float | None:
+    """Parse a Retry-After header value. Returns None on malformed input."""
+    try:
+        return max(0.0, float(int(value)))  # clamp: negative integers are malformed servers
+    except ValueError:
+        pass
+    try:
+        parsed = email.utils.parsedate_to_datetime(value)
+    except (TypeError, ValueError):
+        return None
+    if parsed is None:  # pragma: no cover — parsedate_to_datetime raises rather than returning None in CPython 3.11+
+        return None
+    now = datetime.datetime.now(datetime.UTC)
+    delta = (parsed - now).total_seconds()
+    return max(0.0, delta)
+
+
+class Retry:
+    """Retry middleware. See module docstring for default policy."""
+
+    def __init__(  # noqa: PLR0913 — retry policy has many orthogonal knobs; a dataclass would be worse
+        self,
+        *,
+        max_attempts: int = 3,
+        base_delay: float = 0.1,
+        max_delay: float = 5.0,
+        attempt_timeout: float | None = None,
+        retry_status_codes: frozenset[int] = DEFAULT_RETRY_STATUS_CODES,
+        retry_methods: frozenset[str] = DEFAULT_IDEMPOTENT_METHODS,
+        respect_retry_after: bool = True,
+        budget: RetryBudget | None = None,
+        _sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
+    ) -> None:
+        if max_attempts < 1:
+            raise ValueError(_MAX_ATTEMPTS_INVALID)
+        self.max_attempts = max_attempts
+        self.base_delay = base_delay
+        self.max_delay = max_delay
+        self.attempt_timeout = attempt_timeout
+        self.retry_status_codes = retry_status_codes
+        self.retry_methods = retry_methods
+        self.respect_retry_after = respect_retry_after
+        self.budget = budget if budget is not None else RetryBudget()
+        self._sleep = _sleep
+
+    async def __call__(self, request: httpx2.Request, next: Next) -> httpx2.Response:  # noqa: A002, C901, PLR0912 — complexity budget: 3 error clauses + idempotency gate + budget gate + Retry-After branch + backoff
+        """Process a request through the retry loop. See module docstring."""
+        method_eligible = request.method.upper() in self.retry_methods
+        last_exc: BaseException | None = None
+        last_response: httpx2.Response | None = None
+
+        for attempt in range(self.max_attempts):
+            is_last = attempt + 1 >= self.max_attempts
+            self.budget.deposit()
+            try:
+                if self.attempt_timeout is not None:
+                    async with asyncio.timeout(self.attempt_timeout):
+                        return await next(request)
+                else:
+                    return await next(request)
+            except StatusError as exc:
+                if not method_eligible or exc.response.status_code not in self.retry_status_codes:
+                    raise
+                last_exc = exc
+                last_response = exc.response
+            except (NetworkError, TimeoutError) as exc:
+                if not method_eligible:
+                    raise
+                last_exc = exc
+                last_response = None
+            except builtins.TimeoutError as exc:
+                wrapped = TimeoutError("attempt timed out")
+                wrapped.__cause__ = exc  # set now; the retry path (last_exc = wrapped) has no `from` clause
+                if not method_eligible:
+                    raise wrapped from exc
+                last_exc = wrapped
+                last_response = None
+
+            # ---- retryable failure path
+            if is_last:
+                if last_exc is None:  # pragma: no cover — structural invariant from except branch
+                    msg = "Retry: last_exc unset on final attempt — unreachable"
+                    raise AssertionError(msg)
+                last_exc.add_note(f"httpware: gave up after {attempt + 1} attempts")
+                raise last_exc
+
+            if not self.budget.try_withdraw():
+                raise RetryBudgetExhaustedError(
+                    last_response=last_response,
+                    last_exception=last_exc,
+                    attempts=attempt + 1,
+                ) from last_exc
+
+            retry_after: float | None = None
+            if self.respect_retry_after and last_response is not None:
+                header = last_response.headers.get("Retry-After")
+                if header is not None:
+                    retry_after = _parse_retry_after(header)
+
+            if retry_after is not None:
+                delay = min(retry_after, self.max_delay)
+            else:
+                delay = full_jitter_delay(
+                    attempt,
+                    base_delay=self.base_delay,
+                    max_delay=self.max_delay,
+                )
+            await self._sleep(delay)
+
+        msg = "unreachable"  # pragma: no cover
+        raise AssertionError(msg)  # pragma: no cover
diff --git a/tests/test_backoff.py b/tests/test_backoff.py
new file mode 100644
index 0000000..5b94f42
--- /dev/null
+++ b/tests/test_backoff.py
@@ -0,0 +1,35 @@
+"""Unit tests for the full-jitter backoff helper.
+
+Integration coverage comes from ``tests/test_retry.py`` (Retry middleware drives
+``full_jitter_delay`` per attempt). The pure-function tests here pin the bound
+and the cap independently of the middleware orchestration.
+"""
+
+from httpware.middleware.resilience._backoff import full_jitter_delay
+
+
+BASE_DELAY = 0.1
+MAX_DELAY = 5.0
+
+
+def test_full_jitter_delay_bounded_by_min_of_max_and_exponential() -> None:
+    # attempt_index=0, base=0.1 → ceiling = min(5.0, 0.1*1) = 0.1
+    delay = full_jitter_delay(0, base_delay=BASE_DELAY, max_delay=MAX_DELAY)
+    assert 0.0 <= delay <= BASE_DELAY
+
+
+def test_full_jitter_delay_capped_at_max_delay() -> None:
+    # attempt_index=10, base=0.1 → exp = 102.4 → capped to 5.0
+    delay = full_jitter_delay(10, base_delay=BASE_DELAY, max_delay=MAX_DELAY)
+    assert 0.0 <= delay <= MAX_DELAY
+
+
+def test_full_jitter_delay_uses_injected_random() -> None:
+    # Inject a deterministic mock that returns the upper bound
+    delay = full_jitter_delay(
+        0,
+        base_delay=BASE_DELAY,
+        max_delay=MAX_DELAY,
+        _random_uniform=lambda _lo, hi: hi,
+    )
+    assert delay == BASE_DELAY
diff --git a/tests/test_budget.py b/tests/test_budget.py
new file mode 100644
index 0000000..a94abc3
--- /dev/null
+++ b/tests/test_budget.py
@@ -0,0 +1,104 @@
+"""Unit tests for RetryBudget token-bucket math.
+
+Tests inject a deterministic `_now` callable rather than monkeypatching `time.monotonic`,
+so they cannot be perturbed by other tests sharing the same module.
+"""
+
+import time
+
+from httpware.middleware.resilience.budget import RetryBudget
+
+
+class _Clock:
+    """Mutable clock for deterministic tests. Pass `clock.now` as `_now`."""
+
+    def __init__(self, start: float = 0.0) -> None:
+        self._t = start
+
+    def now(self) -> float:
+        return self._t
+
+    def advance(self, seconds: float) -> None:
+        self._t += seconds
+
+
+def test_defaults_match_spec() -> None:
+    budget = RetryBudget()
+    assert budget._ttl == 10.0  # noqa: SLF001, PLR2004
+    assert budget._min_retries_per_sec == 10.0  # noqa: SLF001, PLR2004
+    assert budget._percent_can_retry == 0.2  # noqa: SLF001, PLR2004
+
+
+def test_floor_permits_min_retries_per_sec_times_ttl_with_zero_deposits() -> None:
+    # floor = min_retries_per_sec * ttl = 10 * 10 = 100 permitted withdrawals
+    clock = _Clock()
+    budget = RetryBudget(ttl=10.0, min_retries_per_sec=10.0, percent_can_retry=0.0, _now=clock.now)
+    permitted = sum(1 for _ in range(101) if budget.try_withdraw())
+    assert permitted == 100  # noqa: PLR2004
+
+
+def test_percent_can_retry_ceiling_with_deposits() -> None:
+    # 1000 deposits * 0.2 = 200 retries permitted (plus floor 100 = 300 total)
+    clock = _Clock()
+    budget = RetryBudget(ttl=10.0, min_retries_per_sec=10.0, percent_can_retry=0.2, _now=clock.now)
+    for _ in range(1000):
+        budget.deposit()
+    permitted = sum(1 for _ in range(500) if budget.try_withdraw())
+    assert permitted == 300  # noqa: PLR2004
+
+
+def test_ttl_expiry_purges_old_deposits() -> None:
+    clock = _Clock()
+    budget = RetryBudget(ttl=1.0, min_retries_per_sec=0.0, percent_can_retry=0.5, _now=clock.now)
+    for _ in range(10):
+        budget.deposit()
+    # 10 deposits * 0.5 = 5 retries available immediately
+    assert budget.try_withdraw() is True
+    # Advance past TTL; deposits expire
+    clock.advance(2.0)
+    # With min_retries_per_sec=0 and no live deposits, no retries permitted
+    assert budget.try_withdraw() is False
+
+
+def test_try_withdraw_returns_false_when_exhausted() -> None:
+    clock = _Clock()
+    budget = RetryBudget(ttl=10.0, min_retries_per_sec=1.0, percent_can_retry=0.0, _now=clock.now)
+    # floor = 1 * 10 = 10 retries
+    for _ in range(10):
+        assert budget.try_withdraw() is True
+    assert budget.try_withdraw() is False
+
+
+def test_deposit_after_exhaustion_does_not_immediately_unblock() -> None:
+    """A single deposit at 20% percent_can_retry contributes 0.2 → int() truncates to 0 → no new retries."""
+    clock = _Clock()
+    budget = RetryBudget(ttl=10.0, min_retries_per_sec=1.0, percent_can_retry=0.2, _now=clock.now)
+    # exhaust the floor (10)
+    for _ in range(10):
+        budget.try_withdraw()
+    assert budget.try_withdraw() is False
+    # one deposit: 1 * 0.2 = 0.2 → int() → 0
+    budget.deposit()
+    assert budget.try_withdraw() is False
+    # 5 more deposits: 6 * 0.2 = 1.2 → int() → 1 new retry permitted
+    for _ in range(5):
+        budget.deposit()
+    assert budget.try_withdraw() is True
+    assert budget.try_withdraw() is False
+
+
+def test_withdrawn_also_expires_after_ttl() -> None:
+    """After TTL passes, prior withdrawals no longer count against the budget."""
+    clock = _Clock()
+    budget = RetryBudget(ttl=1.0, min_retries_per_sec=10.0, percent_can_retry=0.0, _now=clock.now)
+    for _ in range(10):
+        budget.try_withdraw()
+    assert budget.try_withdraw() is False
+    clock.advance(2.0)
+    assert budget.try_withdraw() is True
+
+
+def test_default_now_is_time_monotonic() -> None:
+    """When _now is not passed, the budget uses time.monotonic by default."""
+    budget = RetryBudget()
+    assert budget._now is time.monotonic  # noqa: SLF001
diff --git a/tests/test_budget_props.py b/tests/test_budget_props.py
new file mode 100644
index 0000000..2052fed
--- /dev/null
+++ b/tests/test_budget_props.py
@@ -0,0 +1,115 @@
+"""Hypothesis property tests for RetryBudget.
+
+Properties verified:
+1. `try_withdraw()` never permits more than `floor + int(deposits * percent)` over any window.
+2. After advancing the clock past `ttl`, all prior deposits expire (no retries permitted
+   beyond the floor).
+3. `deposit()` is monotonically non-decreasing in permitted retries (more deposits cannot
+   reduce the budget).
+"""
+
+from collections.abc import Callable
+
+from hypothesis import given, settings
+from hypothesis import strategies as st
+
+from httpware.middleware.resilience.budget import RetryBudget
+
+
+class _Clock:
+    def __init__(self) -> None:
+        self._t = 0.0
+
+    def now(self) -> float:
+        return self._t
+
+    def advance(self, seconds: float) -> None:
+        self._t += seconds
+
+
+def _budget(
+    *,
+    ttl: float,
+    min_retries_per_sec: float,
+    percent_can_retry: float,
+    now: Callable[[], float],
+) -> RetryBudget:
+    return RetryBudget(
+        ttl=ttl,
+        min_retries_per_sec=min_retries_per_sec,
+        percent_can_retry=percent_can_retry,
+        _now=now,
+    )
+
+
+@given(
+    ttl=st.floats(min_value=0.1, max_value=60.0, allow_nan=False, allow_infinity=False),
+    min_rps=st.floats(min_value=0.0, max_value=100.0, allow_nan=False, allow_infinity=False),
+    percent=st.floats(min_value=0.0, max_value=1.0, allow_nan=False, allow_infinity=False),
+    deposits=st.integers(min_value=0, max_value=10_000),
+)
+@settings(max_examples=200, deadline=None)
+def test_try_withdraw_never_exceeds_theoretical_bound(
+    ttl: float,
+    min_rps: float,
+    percent: float,
+    deposits: int,
+) -> None:
+    clock = _Clock()
+    budget = _budget(ttl=ttl, min_retries_per_sec=min_rps, percent_can_retry=percent, now=clock.now)
+    for _ in range(deposits):
+        budget.deposit()
+    floor = int(min_rps * ttl)
+    ceiling = int(deposits * percent) + floor
+    permitted = 0
+    # Try up to ceiling + 10 times to confirm the cap holds.
+    for _ in range(ceiling + 10):
+        if budget.try_withdraw():
+            permitted += 1
+    assert permitted <= ceiling
+
+
+@given(
+    ttl=st.floats(min_value=0.1, max_value=10.0, allow_nan=False, allow_infinity=False),
+    deposits=st.integers(min_value=1, max_value=1000),
+    percent=st.floats(min_value=0.01, max_value=1.0, allow_nan=False, allow_infinity=False),
+)
+@settings(max_examples=100, deadline=None)
+def test_advancing_past_ttl_purges_deposits(ttl: float, deposits: int, percent: float) -> None:
+    clock = _Clock()
+    budget = _budget(ttl=ttl, min_retries_per_sec=0.0, percent_can_retry=percent, now=clock.now)
+    for _ in range(deposits):
+        budget.deposit()
+    # ttl * 2.0 is self-evidently past the TTL window for any positive ttl, and stays
+    # safe across the full strategy range without depending on a hard-coded epsilon.
+    clock.advance(ttl * 2.0)
+    # After purge, no deposits remain; floor is 0 → no retries permitted.
+    assert budget.try_withdraw() is False
+
+
+@given(
+    ttl=st.floats(min_value=0.1, max_value=60.0, allow_nan=False, allow_infinity=False),
+    min_rps=st.floats(min_value=0.0, max_value=100.0, allow_nan=False, allow_infinity=False),
+    percent=st.floats(min_value=0.0, max_value=1.0, allow_nan=False, allow_infinity=False),
+    base_deposits=st.integers(min_value=0, max_value=100),
+    extra_deposits=st.integers(min_value=0, max_value=100),
+)
+@settings(max_examples=50, deadline=None)
+def test_more_deposits_never_decreases_budget(
+    ttl: float,
+    min_rps: float,
+    percent: float,
+    base_deposits: int,
+    extra_deposits: int,
+) -> None:
+    clock = _Clock()
+    budget = _budget(ttl=ttl, min_retries_per_sec=min_rps, percent_can_retry=percent, now=clock.now)
+    for _ in range(base_deposits):
+        budget.deposit()
+    initial_permitted = sum(1 for _ in range(base_deposits + 200) if budget.try_withdraw())
+    # Fresh budget with the same starting deposits + extra
+    budget2 = _budget(ttl=ttl, min_retries_per_sec=min_rps, percent_can_retry=percent, now=clock.now)
+    for _ in range(base_deposits + extra_deposits):
+        budget2.deposit()
+    new_permitted = sum(1 for _ in range(base_deposits + extra_deposits + 200) if budget2.try_withdraw())
+    assert new_permitted >= initial_permitted
diff --git a/tests/test_error_mapping_terminal.py b/tests/test_error_mapping_terminal.py
index 53e3be1..a2a86ad 100644
--- a/tests/test_error_mapping_terminal.py
+++ b/tests/test_error_mapping_terminal.py
@@ -17,6 +17,7 @@
     TimeoutError,  # noqa: A004
     TransportError,
 )
+from httpware.errors import NetworkError
 
 
 def _client_with_handler(handler) -> AsyncClient:  # noqa: ANN001
@@ -81,13 +82,13 @@ def handler(request: httpx2.Request) -> httpx2.Response:  # noqa: ARG001
         await client.send(httpx2.Request("GET", "https://example.test/x"))
 
 
-async def test_httpx2_connect_error_maps_to_transport_error() -> None:
+async def test_httpx2_connect_error_maps_to_network_error() -> None:
     def handler(request: httpx2.Request) -> httpx2.Response:  # noqa: ARG001
         msg = "connect refused"
         raise httpx2.ConnectError(msg)
 
     client = _client_with_handler(handler)
-    with pytest.raises(TransportError, match="connect refused"):
+    with pytest.raises(NetworkError, match="connect refused"):
         await client.send(httpx2.Request("GET", "https://example.test/x"))
 
 
@@ -108,3 +109,29 @@ async def test_send_on_closed_client_raises_transport_error() -> None:
     await underlying.aclose()
     with pytest.raises(TransportError):
         await client.send(httpx2.Request("GET", "https://example.test/x"))
+
+
+async def test_httpx2_decoding_error_maps_to_transport_error() -> None:
+    """Non-transient HTTPError (e.g. DecodingError) maps to bare TransportError, not NetworkError."""
+
+    def handler(request: httpx2.Request) -> httpx2.Response:  # noqa: ARG001
+        msg = "decoding failed"
+        raise httpx2.DecodingError(msg)
+
+    client = _client_with_handler(handler)
+    with pytest.raises(TransportError) as info:
+        await client.send(httpx2.Request("GET", "https://example.test/x"))
+    assert not isinstance(info.value, NetworkError)
+
+
+async def test_httpx2_invalid_url_does_not_map_to_network_error() -> None:
+    """Regression: only transient errors map to NetworkError; InvalidURL stays bare TransportError."""
+
+    def handler(request: httpx2.Request) -> httpx2.Response:  # noqa: ARG001
+        msg = "bad url"
+        raise httpx2.InvalidURL(msg)
+
+    client = _client_with_handler(handler)
+    with pytest.raises(TransportError) as info:
+        await client.send(httpx2.Request("GET", "https://example.test/x"))
+    assert not isinstance(info.value, NetworkError)
diff --git a/tests/test_errors.py b/tests/test_errors.py
index 3cb2011..ffccfa6 100644
--- a/tests/test_errors.py
+++ b/tests/test_errors.py
@@ -14,8 +14,10 @@
     ConflictError,
     ForbiddenError,
     InternalServerError,
+    NetworkError,
     NotFoundError,
     RateLimitedError,
+    RetryBudgetExhaustedError,
     ServerStatusError,
     ServiceUnavailableError,
     StatusError,
@@ -94,6 +96,9 @@ def test_status_error_repr_strips_userinfo() -> None:
 
 
 _NOT_FOUND = 404
+_RETRY_ATTEMPTS_3 = 3
+_RETRY_ATTEMPTS_2 = 2
+_RETRY_ATTEMPTS_5 = 5
 
 
 def test_status_error_pickleable() -> None:
@@ -157,3 +162,49 @@ def test_timeout_error_is_builtin_timeout_error() -> None:
 def test_transport_error_is_client_error() -> None:
     exc = TransportError("connection refused")
     assert isinstance(exc, ClientError)
+
+
+def test_network_error_is_transport_error() -> None:
+    exc = NetworkError("connection refused")
+    assert isinstance(exc, TransportError)
+    assert isinstance(exc, ClientError)
+
+
+def test_retry_budget_exhausted_error_is_client_error() -> None:
+    exc = RetryBudgetExhaustedError(last_response=None, last_exception=None, attempts=_RETRY_ATTEMPTS_3)
+    assert isinstance(exc, ClientError)
+    assert exc.last_response is None
+    assert exc.last_exception is None
+    assert exc.attempts == _RETRY_ATTEMPTS_3
+
+
+def test_retry_budget_exhausted_error_carries_last_response_and_exception() -> None:
+    response = _make_response(503, url="https://example.test/x")
+    inner = RuntimeError("boom")
+    exc = RetryBudgetExhaustedError(last_response=response, last_exception=inner, attempts=_RETRY_ATTEMPTS_2)
+    assert exc.last_response is response
+    assert exc.last_exception is inner
+    assert exc.attempts == _RETRY_ATTEMPTS_2
+
+
+def test_retry_budget_exhausted_error_summary_mentions_attempts() -> None:
+    exc = RetryBudgetExhaustedError(last_response=None, last_exception=None, attempts=_RETRY_ATTEMPTS_5)
+    assert str(exc) == f"retry budget exhausted after {_RETRY_ATTEMPTS_5} attempt(s)"
+
+
+_SERVICE_UNAVAILABLE = 503
+
+
+def test_retry_budget_exhausted_error_pickleable() -> None:
+    response = _make_response(_SERVICE_UNAVAILABLE, url="https://example.test/x")
+    inner = RuntimeError("boom")
+    exc = RetryBudgetExhaustedError(
+        last_response=response,
+        last_exception=inner,
+        attempts=_RETRY_ATTEMPTS_3,
+    )
+    restored = pickle.loads(pickle.dumps(exc))  # noqa: S301
+    assert isinstance(restored, RetryBudgetExhaustedError)
+    assert restored.attempts == _RETRY_ATTEMPTS_3
+    assert restored.last_response is not None
+    assert restored.last_response.status_code == _SERVICE_UNAVAILABLE
diff --git a/tests/test_public_api.py b/tests/test_public_api.py
index a8fca45..92483e7 100644
--- a/tests/test_public_api.py
+++ b/tests/test_public_api.py
@@ -30,8 +30,12 @@ def test_expected_exports() -> None:
     expected = {
         "AsyncClient",
         "Middleware",
+        "NetworkError",
         "Next",
         "ResponseDecoder",
+        "Retry",
+        "RetryBudget",
+        "RetryBudgetExhaustedError",
         "ClientError",
         "TransportError",
         "TimeoutError",
diff --git a/tests/test_retry.py b/tests/test_retry.py
new file mode 100644
index 0000000..a1a2bb5
--- /dev/null
+++ b/tests/test_retry.py
@@ -0,0 +1,443 @@
+"""Tests for the Retry middleware.
+
+Mocks the transport via httpx2.MockTransport; injects a recording `_sleep`
+callable so the suite runs instantly without freezegun.
+"""
+
+import asyncio
+import datetime
+import email.utils
+from collections.abc import Callable
+from http import HTTPStatus
+
+import httpx2
+import pytest
+
+from httpware import AsyncClient, NotFoundError, ServiceUnavailableError, TransportError
+from httpware.errors import NetworkError, RetryBudgetExhaustedError
+from httpware.errors import TimeoutError as HttpwareTimeoutError
+from httpware.middleware.resilience.budget import RetryBudget
+from httpware.middleware.resilience.retry import (
+    DEFAULT_IDEMPOTENT_METHODS,
+    DEFAULT_RETRY_STATUS_CODES,
+    Retry,
+)
+
+
+class _SleepRecorder:
+    def __init__(self) -> None:
+        self.calls: list[float] = []
+
+    async def __call__(self, delay: float) -> None:
+        self.calls.append(delay)
+
+
+class _ResponseSequence:
+    """Mock-transport handler that returns a fixed sequence of responses."""
+
+    def __init__(self, statuses: list[int]) -> None:
+        self._statuses = list(statuses)
+        self.calls: int = 0
+
+    def __call__(self, request: httpx2.Request) -> httpx2.Response:
+        self.calls += 1
+        status = self._statuses.pop(0) if self._statuses else HTTPStatus.OK
+        return httpx2.Response(status, request=request)
+
+
+def _client(handler: Callable[[httpx2.Request], httpx2.Response], *, retry: Retry) -> AsyncClient:
+    transport = httpx2.MockTransport(handler)
+    return AsyncClient(
+        httpx2_client=httpx2.AsyncClient(transport=transport),
+        middleware=[retry],
+    )
+
+
+def test_default_retry_status_codes_match_spec() -> None:
+    assert frozenset({408, 429, 502, 503, 504}) == DEFAULT_RETRY_STATUS_CODES
+
+
+def test_default_idempotent_methods_match_spec() -> None:
+    assert frozenset({"GET", "HEAD", "OPTIONS", "PUT", "DELETE"}) == DEFAULT_IDEMPOTENT_METHODS
+
+
+async def test_succeeds_first_try_no_sleep() -> None:
+    sleeper = _SleepRecorder()
+    handler = _ResponseSequence([HTTPStatus.OK])
+    client = _client(handler, retry=Retry(_sleep=sleeper))
+    response = await client.get("https://example.test/x")
+    assert response.status_code == HTTPStatus.OK
+    assert handler.calls == 1
+    assert sleeper.calls == []
+
+
+async def test_retries_503_then_succeeds() -> None:
+    sleeper = _SleepRecorder()
+    handler = _ResponseSequence([HTTPStatus.SERVICE_UNAVAILABLE, HTTPStatus.OK])
+    client = _client(handler, retry=Retry(_sleep=sleeper, base_delay=0.01, max_delay=0.02))
+    response = await client.get("https://example.test/x")
+    assert response.status_code == HTTPStatus.OK
+    assert handler.calls == 2  # noqa: PLR2004 — "2" is intentional literal in test assertion
+    assert len(sleeper.calls) == 1
+    assert 0.0 <= sleeper.calls[0] <= 0.02  # noqa: PLR2004 — 0.02 matches max_delay literal above
+
+
+async def test_gives_up_after_max_attempts_and_reraises_status_error() -> None:
+    sleeper = _SleepRecorder()
+    handler = _ResponseSequence([HTTPStatus.SERVICE_UNAVAILABLE] * 3)
+    client = _client(handler, retry=Retry(_sleep=sleeper, base_delay=0.01, max_delay=0.02, max_attempts=3))
+    with pytest.raises(ServiceUnavailableError) as info:
+        await client.get("https://example.test/x")
+    assert handler.calls == 3  # noqa: PLR2004 — "3" is intentional literal in test assertion
+    assert len(sleeper.calls) == 2  # noqa: PLR2004 — max_attempts=3 → 2 sleeps between 3 attempts
+    notes = getattr(info.value, "__notes__", [])
+    assert any("gave up after 3 attempts" in note for note in notes)
+
+
+async def test_does_not_retry_non_retryable_status() -> None:
+    sleeper = _SleepRecorder()
+    handler = _ResponseSequence([HTTPStatus.NOT_FOUND])
+    client = _client(handler, retry=Retry(_sleep=sleeper))
+    with pytest.raises(NotFoundError):
+        await client.get("https://example.test/x")
+    assert handler.calls == 1
+    assert sleeper.calls == []
+
+
+async def test_does_not_retry_non_idempotent_methods_by_default() -> None:
+    sleeper = _SleepRecorder()
+    handler = _ResponseSequence([HTTPStatus.SERVICE_UNAVAILABLE])
+    client = _client(handler, retry=Retry(_sleep=sleeper))
+    with pytest.raises(ServiceUnavailableError):
+        await client.post("https://example.test/x", json={"x": 1})
+    assert handler.calls == 1
+    assert sleeper.calls == []
+
+
+async def test_retries_post_when_method_explicitly_included() -> None:
+    sleeper = _SleepRecorder()
+    handler = _ResponseSequence([HTTPStatus.SERVICE_UNAVAILABLE, HTTPStatus.OK])
+    methods = frozenset(DEFAULT_IDEMPOTENT_METHODS | {"POST"})
+    client = _client(
+        handler,
+        retry=Retry(_sleep=sleeper, retry_methods=methods, base_delay=0.01, max_delay=0.02),
+    )
+    response = await client.post("https://example.test/x", json={"x": 1})
+    assert response.status_code == HTTPStatus.OK
+    assert handler.calls == 2  # noqa: PLR2004 — "2" is intentional literal in test assertion
+
+
+async def test_max_attempts_one_means_no_retries() -> None:
+    sleeper = _SleepRecorder()
+    handler = _ResponseSequence([HTTPStatus.SERVICE_UNAVAILABLE])
+    client = _client(handler, retry=Retry(_sleep=sleeper, max_attempts=1))
+    with pytest.raises(ServiceUnavailableError):
+        await client.get("https://example.test/x")
+    assert handler.calls == 1
+    assert sleeper.calls == []
+
+
+def test_max_attempts_zero_rejected() -> None:
+    with pytest.raises(ValueError, match="max_attempts must be >= 1"):
+        Retry(max_attempts=0)
+
+
+async def test_budget_exhausted_raises_retry_budget_exhausted_error() -> None:
+    # NOTE: lives here for coverage of the Retry loop's budget-exhaustion branch.
+    # Task 11 adds the broader budget-gate + sharing tests (carry-through behavior,
+    # last_response / last_exception field population). Do NOT duplicate this test.
+    sleeper = _SleepRecorder()
+    handler = _ResponseSequence([HTTPStatus.SERVICE_UNAVAILABLE, HTTPStatus.SERVICE_UNAVAILABLE])
+    # Budget with zero tolerance: percent_can_retry=0.0, min_retries_per_sec=0.0 → ceiling=0
+    stingy_budget = RetryBudget(percent_can_retry=0.0, min_retries_per_sec=0.0)
+    client = _client(
+        handler,
+        retry=Retry(_sleep=sleeper, budget=stingy_budget, max_attempts=3, base_delay=0.01),
+    )
+    with pytest.raises(RetryBudgetExhaustedError) as info:
+        await client.get("https://example.test/x")
+    assert handler.calls == 1
+    assert info.value.attempts == 1
+    assert sleeper.calls == []
+
+
+async def test_retries_on_network_error() -> None:
+    sleeper = _SleepRecorder()
+    call_count = {"n": 0}
+
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        call_count["n"] += 1
+        if call_count["n"] < 2:  # noqa: PLR2004 — "2" is intentional literal in test assertion
+            msg = "transient"
+            raise httpx2.ConnectError(msg)
+        return httpx2.Response(HTTPStatus.OK, request=request)
+
+    client = _client(handler, retry=Retry(_sleep=sleeper, base_delay=0.01, max_delay=0.02))
+    response = await client.get("https://example.test/x")
+    assert response.status_code == HTTPStatus.OK
+    assert call_count["n"] == 2  # noqa: PLR2004 — "2" is intentional literal in test assertion
+    assert len(sleeper.calls) == 1
+
+
+async def test_retries_on_httpware_timeout_error() -> None:
+    sleeper = _SleepRecorder()
+    call_count = {"n": 0}
+
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        call_count["n"] += 1
+        if call_count["n"] < 2:  # noqa: PLR2004 — "2" is intentional literal in test assertion
+            msg = "read timeout"
+            raise httpx2.ReadTimeout(msg)
+        return httpx2.Response(HTTPStatus.OK, request=request)
+
+    client = _client(handler, retry=Retry(_sleep=sleeper, base_delay=0.01, max_delay=0.02))
+    response = await client.get("https://example.test/x")
+    assert response.status_code == HTTPStatus.OK
+    assert call_count["n"] == 2  # noqa: PLR2004 — "2" is intentional literal in test assertion
+    assert len(sleeper.calls) == 1
+
+
+async def test_does_not_retry_on_bare_transport_error_like_invalid_url() -> None:
+    sleeper = _SleepRecorder()
+
+    def handler(request: httpx2.Request) -> httpx2.Response:  # noqa: ARG001
+        msg = "bad url"
+        raise httpx2.InvalidURL(msg)
+
+    client = _client(handler, retry=Retry(_sleep=sleeper))
+    with pytest.raises(TransportError) as info:
+        await client.get("https://example.test/x")
+    assert not isinstance(info.value, NetworkError)
+    assert sleeper.calls == []
+
+
+async def test_network_error_exhaustion_reraises_with_note() -> None:
+    sleeper = _SleepRecorder()
+
+    def handler(request: httpx2.Request) -> httpx2.Response:  # noqa: ARG001
+        msg = "never works"
+        raise httpx2.ConnectError(msg)
+
+    client = _client(handler, retry=Retry(_sleep=sleeper, max_attempts=2, base_delay=0.01, max_delay=0.02))
+    with pytest.raises(NetworkError) as info:
+        await client.get("https://example.test/x")
+    notes = getattr(info.value, "__notes__", [])
+    assert any("gave up after 2 attempts" in note for note in notes)
+
+
+async def test_does_not_retry_network_error_on_non_idempotent_method() -> None:
+    sleeper = _SleepRecorder()
+    call_count = {"n": 0}
+
+    def handler(request: httpx2.Request) -> httpx2.Response:  # noqa: ARG001
+        call_count["n"] += 1
+        msg = "transient"
+        raise httpx2.ConnectError(msg)
+
+    client = _client(handler, retry=Retry(_sleep=sleeper))
+    with pytest.raises(NetworkError):
+        await client.post("https://example.test/x", json={"x": 1})
+    assert call_count["n"] == 1
+    assert sleeper.calls == []
+
+
+async def test_attempt_timeout_fires_and_retries() -> None:
+    sleeper = _SleepRecorder()
+    call_count = {"n": 0}
+
+    async def handler_async(request: httpx2.Request) -> httpx2.Response:
+        call_count["n"] += 1
+        if call_count["n"] < 2:  # noqa: PLR2004 — "2" is intentional literal in test assertion
+            await asyncio.sleep(1.0)  # exceeds attempt_timeout
+        return httpx2.Response(HTTPStatus.OK, request=request)
+
+    transport = httpx2.MockTransport(handler_async)
+    client = AsyncClient(
+        httpx2_client=httpx2.AsyncClient(transport=transport),
+        middleware=[Retry(_sleep=sleeper, attempt_timeout=0.05, base_delay=0.01, max_delay=0.02)],
+    )
+    response = await client.get("https://example.test/x")
+    # coverage[thread] loses the coroutine frame after asyncio.timeout-induced cancellation.
+    # The assertions DO execute — verified by intentionally breaking them (test fails as
+    # expected). Pragmas mask a tooling limitation, not dead code.
+    assert response.status_code == HTTPStatus.OK  # pragma: no cover
+    assert call_count["n"] == 2  # pragma: no cover  # noqa: PLR2004 — "2" is intentional literal in test assertion
+
+
+async def test_attempt_timeout_exhaustion_raises_httpware_timeout() -> None:
+    sleeper = _SleepRecorder()
+
+    async def slow_handler(request: httpx2.Request) -> httpx2.Response:  # noqa: ARG001
+        await asyncio.sleep(1.0)
+        msg = "should not reach"  # pragma: no cover
+        raise AssertionError(msg)  # pragma: no cover
+
+    transport = httpx2.MockTransport(slow_handler)
+    client = AsyncClient(
+        httpx2_client=httpx2.AsyncClient(transport=transport),
+        middleware=[Retry(_sleep=sleeper, attempt_timeout=0.05, max_attempts=2, base_delay=0.01, max_delay=0.02)],
+    )
+    with pytest.raises(HttpwareTimeoutError) as info:
+        await client.get("https://example.test/x")
+    notes = getattr(info.value, "__notes__", [])
+    assert any("gave up after 2 attempts" in note for note in notes)
+
+
+async def test_attempt_timeout_does_not_retry_on_non_idempotent_method() -> None:
+    sleeper = _SleepRecorder()
+
+    async def slow_handler(request: httpx2.Request) -> httpx2.Response:  # noqa: ARG001
+        await asyncio.sleep(1.0)
+        msg = "should not reach"  # pragma: no cover
+        raise AssertionError(msg)  # pragma: no cover
+
+    transport = httpx2.MockTransport(slow_handler)
+    client = AsyncClient(
+        httpx2_client=httpx2.AsyncClient(transport=transport),
+        middleware=[Retry(_sleep=sleeper, attempt_timeout=0.05)],
+    )
+    with pytest.raises(HttpwareTimeoutError):
+        await client.post("https://example.test/x", json={"x": 1})
+    assert sleeper.calls == []  # not retried
+
+
+class _ResponseSequenceWithHeaders:
+    """Mock handler that returns (status, headers) tuples in sequence."""
+
+    def __init__(self, responses: list[tuple[int, dict[str, str]]]) -> None:
+        self._responses = list(responses)
+        self.calls = 0
+
+    def __call__(self, request: httpx2.Request) -> httpx2.Response:
+        self.calls += 1
+        status, headers = self._responses.pop(0)
+        return httpx2.Response(status, request=request, headers=headers)
+
+
+async def test_retry_after_seconds_overrides_backoff() -> None:
+    sleeper = _SleepRecorder()
+    handler = _ResponseSequenceWithHeaders(
+        [
+            (HTTPStatus.SERVICE_UNAVAILABLE, {"Retry-After": "2"}),
+            (HTTPStatus.OK, {}),
+        ]
+    )
+    client = _client(handler, retry=Retry(_sleep=sleeper, base_delay=0.01, max_delay=5.0))
+    response = await client.get("https://example.test/x")
+    assert response.status_code == HTTPStatus.OK
+    assert sleeper.calls == [2.0]
+
+
+async def test_retry_after_http_date_overrides_backoff() -> None:
+    sleeper = _SleepRecorder()
+    future = datetime.datetime.now(datetime.UTC) + datetime.timedelta(seconds=3)
+    http_date = email.utils.format_datetime(future, usegmt=True)
+    handler = _ResponseSequenceWithHeaders(
+        [
+            (HTTPStatus.SERVICE_UNAVAILABLE, {"Retry-After": http_date}),
+            (HTTPStatus.OK, {}),
+        ]
+    )
+    client = _client(handler, retry=Retry(_sleep=sleeper, base_delay=0.01, max_delay=10.0))
+    response = await client.get("https://example.test/x")
+    assert response.status_code == HTTPStatus.OK
+    assert len(sleeper.calls) == 1
+    assert 2.0 <= sleeper.calls[0] <= 4.0  # noqa: PLR2004 — ~3 seconds, with clock-skew tolerance
+
+
+async def test_retry_after_capped_at_max_delay() -> None:
+    sleeper = _SleepRecorder()
+    handler = _ResponseSequenceWithHeaders(
+        [
+            (HTTPStatus.SERVICE_UNAVAILABLE, {"Retry-After": "9999"}),
+            (HTTPStatus.OK, {}),
+        ]
+    )
+    client = _client(handler, retry=Retry(_sleep=sleeper, base_delay=0.01, max_delay=2.5))
+    await client.get("https://example.test/x")
+    assert sleeper.calls == [2.5]
+
+
+async def test_malformed_retry_after_falls_back_to_backoff() -> None:
+    sleeper = _SleepRecorder()
+    handler = _ResponseSequenceWithHeaders(
+        [
+            (HTTPStatus.SERVICE_UNAVAILABLE, {"Retry-After": "not-a-number"}),
+            (HTTPStatus.OK, {}),
+        ]
+    )
+    client = _client(handler, retry=Retry(_sleep=sleeper, base_delay=0.01, max_delay=0.05))
+    await client.get("https://example.test/x")
+    assert len(sleeper.calls) == 1
+    assert 0.0 <= sleeper.calls[0] <= 0.05  # noqa: PLR2004 — 0.05 matches max_delay literal above
+
+
+async def test_respect_retry_after_false_ignores_header() -> None:
+    sleeper = _SleepRecorder()
+    handler = _ResponseSequenceWithHeaders(
+        [
+            (HTTPStatus.SERVICE_UNAVAILABLE, {"Retry-After": "5"}),
+            (HTTPStatus.OK, {}),
+        ]
+    )
+    client = _client(
+        handler,
+        retry=Retry(_sleep=sleeper, respect_retry_after=False, base_delay=0.01, max_delay=0.02),
+    )
+    await client.get("https://example.test/x")
+    assert len(sleeper.calls) == 1
+    assert 0.0 <= sleeper.calls[0] <= 0.02  # noqa: PLR2004 — backoff range, not 5
+
+
+def _zero_budget() -> RetryBudget:
+    """Return a budget that always refuses withdrawal (floor=0, percent=0)."""
+    return RetryBudget(ttl=10.0, min_retries_per_sec=0.0, percent_can_retry=0.0)
+
+
+async def test_budget_exhausted_raises_specific_exception() -> None:
+    sleeper = _SleepRecorder()
+    handler = _ResponseSequence([HTTPStatus.SERVICE_UNAVAILABLE, HTTPStatus.OK])
+    client = _client(
+        handler,
+        retry=Retry(_sleep=sleeper, budget=_zero_budget(), base_delay=0.01, max_delay=0.02),
+    )
+    with pytest.raises(RetryBudgetExhaustedError) as info:
+        await client.get("https://example.test/x")
+    assert info.value.attempts == 1  # one attempt made, budget refused before retry
+    assert info.value.last_response is not None
+    assert info.value.last_response.status_code == HTTPStatus.SERVICE_UNAVAILABLE
+    assert isinstance(info.value.last_exception, ServiceUnavailableError)
+
+
+async def test_budget_exhausted_on_network_error_carries_exception_not_response() -> None:
+    sleeper = _SleepRecorder()
+
+    def handler(request: httpx2.Request) -> httpx2.Response:  # noqa: ARG001
+        msg = "transient"
+        raise httpx2.ConnectError(msg)
+
+    client = _client(
+        handler,
+        retry=Retry(_sleep=sleeper, budget=_zero_budget(), base_delay=0.01, max_delay=0.02),
+    )
+    with pytest.raises(RetryBudgetExhaustedError) as info:
+        await client.get("https://example.test/x")
+    assert info.value.last_response is None
+    assert isinstance(info.value.last_exception, NetworkError)
+
+
+async def test_default_budget_is_fresh_per_instance() -> None:
+    r1 = Retry()
+    r2 = Retry()
+    assert r1.budget is not r2.budget
+
+
+async def test_explicit_budget_shared_across_retry_instances() -> None:
+    shared = RetryBudget(ttl=10.0, min_retries_per_sec=1.0, percent_can_retry=0.0)
+    r1 = Retry(budget=shared)
+    r2 = Retry(budget=shared)
+    assert r1.budget is r2.budget
+    # 10 retries total before exhaustion (floor=10)
+    for _ in range(10):
+        assert shared.try_withdraw() is True
+    assert shared.try_withdraw() is False
diff --git a/tests/test_retry_props.py b/tests/test_retry_props.py
new file mode 100644
index 0000000..c4a8c14
--- /dev/null
+++ b/tests/test_retry_props.py
@@ -0,0 +1,171 @@
+"""Hypothesis property tests for Retry.
+
+Properties verified:
+1. Total attempts never exceed max_attempts.
+2. Total sleep time never exceeds max_attempts * max_delay.
+3. Non-retryable statuses (NOT in retry_status_codes) cause exactly one attempt.
+4. Non-idempotent methods (NOT in retry_methods) cause exactly one attempt,
+   regardless of response status.
+"""
+
+from http import HTTPStatus
+
+import httpx2
+from hypothesis import given, settings
+from hypothesis import strategies as st
+
+from httpware import AsyncClient
+from httpware.middleware.resilience.budget import RetryBudget
+from httpware.middleware.resilience.retry import (
+    DEFAULT_IDEMPOTENT_METHODS,
+    DEFAULT_RETRY_STATUS_CODES,
+    Retry,
+)
+
+
+class _SleepRecorder:
+    def __init__(self) -> None:
+        self.calls: list[float] = []
+
+    async def __call__(self, delay: float) -> None:
+        self.calls.append(delay)
+
+
+def _always_status(status: int) -> httpx2.MockTransport:
+    return httpx2.MockTransport(lambda req: httpx2.Response(status, request=req))
+
+
+_RETRYABLE_STATUS_STRATEGY = st.sampled_from(sorted(DEFAULT_RETRY_STATUS_CODES))
+_NON_RETRYABLE_STATUS_STRATEGY = st.sampled_from(
+    [
+        HTTPStatus.BAD_REQUEST,
+        HTTPStatus.UNAUTHORIZED,
+        HTTPStatus.NOT_FOUND,
+        HTTPStatus.CONFLICT,
+    ]
+)
+_IDEMPOTENT_METHODS = st.sampled_from(sorted(DEFAULT_IDEMPOTENT_METHODS))
+_NON_IDEMPOTENT_METHODS = st.sampled_from(["POST", "PATCH"])
+
+
+@given(
+    max_attempts=st.integers(min_value=1, max_value=5),
+    status=_RETRYABLE_STATUS_STRATEGY,
+    method=_IDEMPOTENT_METHODS,
+)
+@settings(max_examples=50, deadline=None)
+async def test_total_attempts_never_exceeds_max_attempts(
+    max_attempts: int,
+    status: int,
+    method: str,
+) -> None:
+    sleeper = _SleepRecorder()
+    call_count = {"n": 0}
+
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        call_count["n"] += 1
+        return httpx2.Response(status, request=request)
+
+    transport = httpx2.MockTransport(handler)
+    client = AsyncClient(
+        httpx2_client=httpx2.AsyncClient(transport=transport),
+        middleware=[
+            Retry(
+                _sleep=sleeper,
+                max_attempts=max_attempts,
+                base_delay=0.001,
+                max_delay=0.002,
+                budget=RetryBudget(ttl=60.0, min_retries_per_sec=1000.0),
+            )
+        ],
+    )
+    try:  # noqa: SIM105 — contextlib.suppress can't be used in async Hypothesis tests
+        await client.request(method, "https://example.test/x")
+    except Exception:  # noqa: BLE001, S110 — we only care about call count
+        pass
+    assert call_count["n"] <= max_attempts
+
+
+@given(
+    max_attempts=st.integers(min_value=1, max_value=5),
+    base_delay=st.floats(min_value=0.001, max_value=0.01),
+    max_delay=st.floats(min_value=0.001, max_value=0.05),
+)
+@settings(max_examples=30, deadline=None)
+async def test_total_sleep_never_exceeds_max_attempts_times_max_delay(
+    max_attempts: int,
+    base_delay: float,
+    max_delay: float,
+) -> None:
+    sleeper = _SleepRecorder()
+    transport = _always_status(HTTPStatus.SERVICE_UNAVAILABLE)
+    client = AsyncClient(
+        httpx2_client=httpx2.AsyncClient(transport=transport),
+        middleware=[
+            Retry(
+                _sleep=sleeper,
+                max_attempts=max_attempts,
+                base_delay=base_delay,
+                max_delay=max_delay,
+                budget=RetryBudget(ttl=60.0, min_retries_per_sec=1000.0),
+            )
+        ],
+    )
+    try:  # noqa: SIM105 — contextlib.suppress can't be used in async Hypothesis tests
+        await client.get("https://example.test/x")
+    except Exception:  # noqa: BLE001, S110
+        pass
+    total = sum(sleeper.calls)
+    assert total <= max_attempts * max_delay + 1e-9
+
+
+@given(
+    status=_NON_RETRYABLE_STATUS_STRATEGY,
+    method=_IDEMPOTENT_METHODS,
+)
+@settings(max_examples=30, deadline=None)
+async def test_non_retryable_status_causes_one_attempt(status: int, method: str) -> None:
+    sleeper = _SleepRecorder()
+    call_count = {"n": 0}
+
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        call_count["n"] += 1
+        return httpx2.Response(status, request=request)
+
+    transport = httpx2.MockTransport(handler)
+    client = AsyncClient(
+        httpx2_client=httpx2.AsyncClient(transport=transport),
+        middleware=[Retry(_sleep=sleeper, max_attempts=3, base_delay=0.001, max_delay=0.002)],
+    )
+    try:  # noqa: SIM105 — contextlib.suppress can't be used in async Hypothesis tests
+        await client.request(method, "https://example.test/x")
+    except Exception:  # noqa: BLE001, S110
+        pass
+    assert call_count["n"] == 1
+    assert sleeper.calls == []
+
+
+@given(
+    status=_RETRYABLE_STATUS_STRATEGY,
+    method=_NON_IDEMPOTENT_METHODS,
+)
+@settings(max_examples=30, deadline=None)
+async def test_non_idempotent_method_causes_one_attempt(status: int, method: str) -> None:
+    sleeper = _SleepRecorder()
+    call_count = {"n": 0}
+
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        call_count["n"] += 1
+        return httpx2.Response(status, request=request)
+
+    transport = httpx2.MockTransport(handler)
+    client = AsyncClient(
+        httpx2_client=httpx2.AsyncClient(transport=transport),
+        middleware=[Retry(_sleep=sleeper, max_attempts=3, base_delay=0.001, max_delay=0.002)],
+    )
+    try:  # noqa: SIM105 — contextlib.suppress can't be used in async Hypothesis tests
+        await client.request(method, "https://example.test/x")
+    except Exception:  # noqa: BLE001, S110
+        pass
+    assert call_count["n"] == 1
+    assert sleeper.calls == []