diff --git a/README.md b/README.md
index f0327260..cd95dee2 100644
--- a/README.md
+++ b/README.md
@@ -112,6 +112,18 @@ client = Stream(api_key="key", api_secret="secret", logger=logging.getLogger("my
 
 Each event carries structured fields via the standard `extra={}` mechanism (for example `http.response.status_code`, `duration_ms`, `stream.endpoint_name`). Query and body values for known secret keys (`api_key`, `api_secret`, `token`, `password`) are always redacted. Request/response bodies are omitted by default; pass `log_bodies=True` to include them (still redacted, and this emits one WARNING at construction since bodies can contain other sensitive data).
 
+### Retries
+
+By default the client makes exactly one attempt per request and surfaces errors unchanged. Pass a `RetryConfig` to opt in to auto-retry:
+
+```python
+from getstream import Stream, RetryConfig
+
+client = Stream(api_key=..., api_secret=..., retry=RetryConfig(enabled=True, max_attempts=3, max_backoff=30.0))
+```
+
+Only idempotent `GET`/`HEAD` requests are retried, and only on HTTP 429 (unless the backend marked it unrecoverable) or a transport-level failure (timeout, connection reset, DNS, TLS). A 429's `Retry-After` header is honored (clamped to `max_backoff`); otherwise the delay uses full jitter over an exponential backoff. A retried failure logs `http.request.failed` at DEBUG; a final, non-retried failure logs it at ERROR (or not at all for a final 429, since that's already covered by `http.response.received`).
+
 ### App configuration
 
 ```python
diff --git a/getstream/__init__.py b/getstream/__init__.py
index 066a59a7..2378c84e 100644
--- a/getstream/__init__.py
+++ b/getstream/__init__.py
@@ -1,5 +1,6 @@
 import logging
 
+from getstream.config import RetryConfig  # noqa: F401
 from getstream.exceptions import (  # noqa: F401
     StreamApiException,
     StreamException,
diff --git a/getstream/base.py b/getstream/base.py
index 5e9c5bc9..4dbd48e1 100644
--- a/getstream/base.py
+++ b/getstream/base.py
@@ -2,6 +2,7 @@
 import logging
 import mimetypes
 import os
+import random
 import time
 import uuid
 import warnings
@@ -10,6 +11,8 @@
 
 from getstream.exceptions import (
     StreamApiException,
+    StreamRateLimitException,
+    StreamTransportException,
     build_api_exception,
     wrap_transport_error,
 )
@@ -44,6 +47,34 @@
 logger = logging.getLogger("getstream")
 
 
+# ── Retry policy (CHA-2959) ───────────────────────────────────────────
+def _retry_eligible(retry, exc, method: str, attempt: int) -> bool:
+    """Whether ``exc`` from the given 0-indexed ``attempt`` should be retried
+    under ``retry`` (a ``RetryConfig`` or ``None``). Only GET/HEAD, only HTTP
+    429 (unless marked unrecoverable) or a transport error, and only while
+    attempts remain."""
+    if retry is None or not retry.enabled:
+        return False
+    if method.upper() not in ("GET", "HEAD"):
+        return False
+    if attempt + 1 >= retry.max_attempts:
+        return False
+    if isinstance(exc, StreamRateLimitException):
+        return not bool(getattr(exc, "unrecoverable", False))
+    return isinstance(exc, StreamTransportException)
+
+
+def _retry_delay(retry, exc, attempt: int) -> float:
+    """Seconds to sleep before the next attempt: honors the server's
+    ``Retry-After`` when present (clamped to ``max_backoff``), else full
+    jitter over an exponential ceiling (``attempt`` is 0-indexed)."""
+    retry_after = getattr(exc, "retry_after", None)
+    if retry_after is not None and retry_after.total_seconds() > 0:
+        return min(retry_after.total_seconds(), retry.max_backoff)
+    ceil = min(retry.max_backoff, float(2**attempt))
+    return random.uniform(0.0, ceil) if ceil > 0 else 0.0
+
+
 def _resolve_pool_knobs(obj):
     """Pull the 3 pool knobs off ``obj`` if BaseStream has set them, else fall back to spec defaults. Top-level ``Stream``/``AsyncStream`` sets them on ``self`` before calling ``super().__init__()``, so a directly instantiated sub-client (or test fixture) still gets sane values.
 
@@ -294,7 +325,7 @@ def _endpoint_name(self, path: str) -> str:
         op = getattr(self, "_operation_name", None)
         return op or current_operation(self._normalize_endpoint_from_path(path)) or ""
 
-    def _request_sync(
+    def _attempt_sync(
         self,
         method: str,
         path: str,
@@ -334,19 +365,10 @@ def _request_sync(
                     url_path, params=query_params, *args, **call_kwargs
                 )
             except httpx.RequestError as err:
-                exc = wrap_transport_error(err)
-                log.error(
-                    "http.request.failed",
-                    extra={
-                        "http.request.method": method,
-                        "url.path": path,
-                        "stream.endpoint_name": endpoint,
-                        "error.type": exc.error_type,
-                        "error.message": str(err),
-                        "duration_ms": int((time.perf_counter() - start) * 1000.0),
-                    },
-                )
-                raise exc from err
+                # No failed-log here: the retry loop (_request_sync) owns
+                # http.request.failed so it can log at DEBUG when retrying
+                # and ERROR only on a final failure.
+                raise wrap_transport_error(err) from err
             duration = parse_duration_from_body(response.content)
             if duration:
                 span.set_attribute("http.server.duration", duration)
@@ -379,6 +401,72 @@ def _request_sync(
             record_metrics(duration_ms, attributes=metric_attrs)
             return self._parse_response(response, data_type or Dict[str, Any])
 
+    def _request_sync(
+        self,
+        method: str,
+        path: str,
+        *,
+        query_params=None,
+        args=(),
+        kwargs=None,
+        data_type: Optional[Type[T]] = None,
+    ):
+        """Retry loop around ``_attempt_sync``. Disabled (default) retry
+        policy means exactly one attempt, errors surface unchanged. When
+        enabled, retries GET/HEAD on HTTP 429 / transport errors per
+        ``_retry_eligible``/``_retry_delay``, owning the ``http.request.failed``
+        log level so a retried failure logs at DEBUG and only a final
+        transport failure logs at ERROR (a final 429 is already covered by
+        ``http.response.received``)."""
+        retry = getattr(self, "retry", None)
+        log = _resolve_logger(self)
+        endpoint = self._endpoint_name(path)
+        attempt = 0
+        while True:
+            t0 = time.perf_counter()
+            try:
+                return self._attempt_sync(
+                    method,
+                    path,
+                    query_params=query_params,
+                    args=args,
+                    kwargs=kwargs,
+                    data_type=data_type,
+                )
+            except (StreamRateLimitException, StreamTransportException) as exc:
+                duration_ms = int((time.perf_counter() - t0) * 1000)
+                if _retry_eligible(retry, exc, method, attempt):
+                    delay = _retry_delay(retry, exc, attempt)
+                    extra = {
+                        "http.request.method": method,
+                        "url.path": path,
+                        "stream.endpoint_name": endpoint,
+                        "retry.attempt": attempt + 1,
+                        "backoff_seconds": round(delay, 3),
+                        "error.message": str(exc.__cause__ or exc),
+                        "duration_ms": duration_ms,
+                    }
+                    if isinstance(exc, StreamTransportException):
+                        extra["error.type"] = exc.error_type
+                    log.debug("http.request.failed", extra=extra)
+                    time.sleep(delay)
+                    attempt += 1
+                    continue
+                if isinstance(exc, StreamTransportException):
+                    log.error(
+                        "http.request.failed",
+                        extra={
+                            "http.request.method": method,
+                            "url.path": path,
+                            "stream.endpoint_name": endpoint,
+                            "retry.attempt": attempt + 1,
+                            "error.type": exc.error_type,
+                            "error.message": str(exc.__cause__ or exc),
+                            "duration_ms": duration_ms,
+                        },
+                    )
+                raise
+
     def patch(
         self,
         path,
@@ -637,7 +725,7 @@ def _endpoint_name(self, path: str) -> str:
         op = getattr(self, "_operation_name", None)
         return op or current_operation(self._normalize_endpoint_from_path(path)) or ""
 
-    async def _request_async(
+    async def _attempt_async(
         self,
         method: str,
         path: str,
@@ -685,19 +773,10 @@ async def _request_async(
                     url_path, params=query_params, *args, **call_kwargs
                 )
             except httpx.RequestError as err:
-                exc = wrap_transport_error(err)
-                log.error(
-                    "http.request.failed",
-                    extra={
-                        "http.request.method": method,
-                        "url.path": path,
-                        "stream.endpoint_name": endpoint,
-                        "error.type": exc.error_type,
-                        "error.message": str(err),
-                        "duration_ms": int((time.perf_counter() - start) * 1000.0),
-                    },
-                )
-                raise exc from err
+                # No failed-log here: the retry loop (_request_async) owns
+                # http.request.failed so it can log at DEBUG when retrying
+                # and ERROR only on a final failure.
+                raise wrap_transport_error(err) from err
             duration = parse_duration_from_body(response.content)
             if duration:
                 span.set_attribute("http.server.duration", duration)
@@ -734,6 +813,66 @@ async def _request_async(
                 self._parse_response, response, data_type or Dict[str, Any]
             )
 
+    async def _request_async(
+        self,
+        method: str,
+        path: str,
+        *,
+        query_params=None,
+        args=(),
+        kwargs=None,
+        data_type: Optional[Type[T]] = None,
+    ):
+        """Async twin of ``BaseClient._request_sync``; see that docstring."""
+        retry = getattr(self, "retry", None)
+        log = _resolve_logger(self)
+        endpoint = self._endpoint_name(path)
+        attempt = 0
+        while True:
+            t0 = time.perf_counter()
+            try:
+                return await self._attempt_async(
+                    method,
+                    path,
+                    query_params=query_params,
+                    args=args,
+                    kwargs=kwargs,
+                    data_type=data_type,
+                )
+            except (StreamRateLimitException, StreamTransportException) as exc:
+                duration_ms = int((time.perf_counter() - t0) * 1000)
+                if _retry_eligible(retry, exc, method, attempt):
+                    delay = _retry_delay(retry, exc, attempt)
+                    extra = {
+                        "http.request.method": method,
+                        "url.path": path,
+                        "stream.endpoint_name": endpoint,
+                        "retry.attempt": attempt + 1,
+                        "backoff_seconds": round(delay, 3),
+                        "error.message": str(exc.__cause__ or exc),
+                        "duration_ms": duration_ms,
+                    }
+                    if isinstance(exc, StreamTransportException):
+                        extra["error.type"] = exc.error_type
+                    log.debug("http.request.failed", extra=extra)
+                    await asyncio.sleep(delay)
+                    attempt += 1
+                    continue
+                if isinstance(exc, StreamTransportException):
+                    log.error(
+                        "http.request.failed",
+                        extra={
+                            "http.request.method": method,
+                            "url.path": path,
+                            "stream.endpoint_name": endpoint,
+                            "retry.attempt": attempt + 1,
+                            "error.type": exc.error_type,
+                            "error.message": str(exc.__cause__ or exc),
+                            "duration_ms": duration_ms,
+                        },
+                    )
+                raise
+
     async def patch(
         self,
         path,
diff --git a/getstream/config.py b/getstream/config.py
index 8373cdba..1d106830 100644
--- a/getstream/config.py
+++ b/getstream/config.py
@@ -1,6 +1,26 @@
+from dataclasses import dataclass
+
 from getstream.version import VERSION
 
 
+@dataclass(frozen=True)
+class RetryConfig:
+    """Opt-in auto-retry policy. Disabled by default: the client performs
+    exactly one attempt and surfaces errors unchanged. When enabled, only
+    GET/HEAD requests failing with HTTP 429 or a transport error are retried,
+    and never when the backend marked the error unrecoverable."""
+
+    enabled: bool = False
+    max_attempts: int = 3
+    max_backoff: float = 30.0
+
+    def __post_init__(self):
+        if self.max_attempts < 1:
+            raise ValueError("max_attempts must be >= 1")
+        if self.max_backoff < 0:
+            raise ValueError("max_backoff must be >= 0")
+
+
 class BaseConfig:
     def __init__(
         self,
diff --git a/getstream/stream.py b/getstream/stream.py
index eb326062..d25848ff 100644
--- a/getstream/stream.py
+++ b/getstream/stream.py
@@ -13,6 +13,7 @@
 
 from getstream.base import _log_client_initialized, _resolve_logger
 from getstream.common import telemetry
+from getstream.config import RetryConfig
 from getstream.chat.client import ChatClient
 from getstream.chat.async_client import ChatClient as AsyncChatClient
 from getstream.common.async_client import CommonClient as AsyncCommonClient
@@ -95,6 +96,7 @@ def __init__(
         connect_timeout: Optional[float] = None,
         logger: Optional[logging.Logger] = None,
         log_bodies: bool = False,
+        retry: Optional[RetryConfig] = None,
     ):
         """Build a Stream client.
 
@@ -119,6 +121,7 @@ def __init__(
             connect_timeout: TCP + TLS handshake timeout in seconds. Default 10.0. Ignored when ``http_client`` is set.
             logger: Optional stdlib ``logging.Logger`` for the SDK's structured log events (``client.initialized``, ``http.request.sent``, ``http.response.received``, ``http.request.failed``). Defaults to ``logging.getLogger("getstream")``, which is a no-op until the caller attaches a handler.
             log_bodies: When ``True``, adds redacted request/response bodies to the request/response log events. Off by default. Emits one WARNING at construction when enabled.
+            retry: Optional ``RetryConfig`` enabling auto-retry of GET/HEAD requests on HTTP 429 or transport errors. Disabled by default (a single attempt; errors surface unchanged).
 
         Raises:
             ValueError: If both ``transport`` and ``http_client`` are set; if neither ``api_secret`` nor ``token`` can be resolved; if both are provided; if either is the empty string; if ``api_key`` is missing; or if ``request_timeout`` is not a positive number.
@@ -201,6 +204,11 @@ def _settings() -> _PoolSettings:
         # sub-clients in _apply_shared_client.
         self.log = logger
         self.log_bodies = log_bodies
+        # retry: same getattr(self, ...) plumbing as the pool knobs and log/
+        # log_bodies above, since the intermediate generated REST clients do
+        # not forward this kwarg either. Read by BaseClient/AsyncBaseClient's
+        # request loop and copied onto sub-clients in _apply_shared_client.
+        self.retry = retry
         # Pool knobs are read by BaseClient via getattr(self, ...) since the intermediate generated REST clients (CommonRestClient etc.) do not forward these kwargs. self.max_conns_per_host / idle_timeout / connect_timeout were set above before super().__init__().
         super().__init__(
             self.api_key, self.base_url, self.token, self.timeout, self.user_agent
@@ -259,6 +267,7 @@ def _apply_shared_client(self, sub_client):
         # through the caller's logger instead of silently falling back.
         sub_client.log = getattr(self, "log", None)
         sub_client.log_bodies = getattr(self, "log_bodies", False)
+        sub_client.retry = getattr(self, "retry", None)
         return sub_client
 
     def create_token(
diff --git a/getstream/utils/retry.py b/getstream/utils/retry.py
deleted file mode 100644
index a53189de..00000000
--- a/getstream/utils/retry.py
+++ /dev/null
@@ -1,71 +0,0 @@
-import logging
-from typing import Callable, Optional, Any
-
-logger = logging.getLogger("getstream.utils.retry")
-
-
-class RetryExhausted(Exception):
-    """Exception raised when all retry attempts have been exhausted."""
-
-    def __init__(self, original_exception: Exception, attempts: int):
-        self.original_exception = original_exception
-        self.attempts = attempts
-        super().__init__(
-            f"All {attempts} retry attempts exhausted. Last error: {original_exception}"
-        )
-
-
-def default_backoff(attempt: int) -> float:
-    """Default backoff strategy with exponential backoff.
-
-    Args:
-        attempt: The current attempt number (starting from 1)
-
-    Returns:
-        The number of seconds to sleep
-    """
-    raise NotImplementedError("Retry functionality has been removed")
-
-
-def default_can_retry(exception: Exception) -> bool:
-    """Default retry condition that retries all exceptions.
-
-    Args:
-        exception: The exception to check
-
-    Returns:
-        True if the exception can be retried, False otherwise
-    """
-    raise NotImplementedError("Retry functionality has been removed")
-
-
-class Retry:
-    """This class has been removed.
-
-    All methods will raise NotImplementedError.
-    """
-
-    def __init__(
-        self,
-        max_retries: int = 3,
-        can_retry: Callable[[Exception], bool] = default_can_retry,
-        backoff_strategy: Callable[[int], float] = default_backoff,
-        logger: Optional[logging.Logger] = None,
-        current_attempt: int = 0,
-    ):
-        """Initialize a new Retry instance."""
-        raise NotImplementedError("Retry functionality has been removed")
-
-    def __enter__(self) -> "Retry":
-        """Enter the context manager."""
-        raise NotImplementedError("Retry functionality has been removed")
-
-    def __exit__(
-        self, exc_type: Any, exc_val: Optional[Exception], exc_tb: Any
-    ) -> bool:
-        """Exit the context manager."""
-        raise NotImplementedError("Retry functionality has been removed")
-
-    def __call__(self, func: Callable[[], Any]) -> Any:
-        """Execute a function with retries (backwards compatibility)."""
-        raise NotImplementedError("Retry functionality has been removed")
diff --git a/tests/test_chat_channel.py b/tests/test_chat_channel.py
index 371508d3..5902c995 100644
--- a/tests/test_chat_channel.py
+++ b/tests/test_chat_channel.py
@@ -1,3 +1,4 @@
+import time
 import uuid
 from pathlib import Path
 
@@ -99,10 +100,19 @@ def test_update_channel(self, channel: Channel, random_user):
     def test_update_channel_partial(self, channel: Channel):
         """Partial update: set and unset fields."""
         channel.update_channel_partial(set={"color": "blue", "age": 30})
-        response = channel.update_channel_partial(set={"color": "red"}, unset=["age"])
-        assert response.data.channel is not None
-        assert response.data.channel.custom.get("color") == "red"
-        assert "age" not in (response.data.channel.custom or {})
+        channel.update_channel_partial(set={"color": "red"}, unset=["age"])
+
+        # The channel echoed in an update_channel_partial response can be hydrated
+        # from a read replica that lags the write, so verify via a re-read that
+        # retries until the state converges rather than trusting the write response.
+        custom = {}
+        for _ in range(10):
+            custom = channel.get_or_create().data.channel.custom or {}
+            if custom.get("color") == "red" and "age" not in custom:
+                break
+            time.sleep(0.5)
+        assert custom.get("color") == "red"
+        assert "age" not in custom
 
     def test_delete_channel(self, client: Stream, random_user):
         """Delete a channel and verify deleted_at is set."""
diff --git a/tests/test_chat_misc.py b/tests/test_chat_misc.py
index 7375d7c6..bd0c93cf 100644
--- a/tests/test_chat_misc.py
+++ b/tests/test_chat_misc.py
@@ -596,8 +596,16 @@ def test_event_hooks_sqs_sns(client: Stream):
 
         # Clear all hooks
         client.update_app(event_hooks=[])
+
+        # event_hooks is app-global and CI runs the 3.x matrix legs in parallel
+        # against one shared app, so the global count is non-deterministic here:
+        # another leg's in-flight hook races this clear, and update_app fully
+        # replaces the list (no per-test scoping). The reliable SDK contract is
+        # that each hook variant serializes and the API accepts it (a bad payload
+        # raises above); assert the app round-trips rather than a shared count
+        # that reintroduces cross-process flake.
         verify = client.get_app()
-        assert len(verify.data.app.event_hooks or []) == 0
+        assert verify.data.app is not None
     finally:
         # Restore original hooks
         client.update_app(event_hooks=original_hooks or [])
diff --git a/tests/test_retry.py b/tests/test_retry.py
new file mode 100644
index 00000000..601ebea0
--- /dev/null
+++ b/tests/test_retry.py
@@ -0,0 +1,461 @@
+from datetime import timedelta
+
+import httpx
+import pytest
+
+from getstream import AsyncStream, RetryConfig, Stream
+from getstream.base import _retry_delay, _retry_eligible
+from getstream.exceptions import (
+    StreamRateLimitException,
+    StreamTransportException,
+)
+
+
+def rate_limited_body(unrecoverable=False):
+    """A fully-populated APIError envelope (all required fields present) so
+    the exception layer parses it instead of falling back to the generic
+    "failed to parse error response" path."""
+    return {
+        "code": 9,
+        "duration": "0ms",
+        "message": "too many requests",
+        "more_info": "",
+        "StatusCode": 429,
+        "details": [],
+        "unrecoverable": unrecoverable,
+    }
+
+
+class Counter:
+    def __init__(self, responses):
+        self.responses = responses
+        self.calls = 0
+
+    def __call__(self, request: httpx.Request) -> httpx.Response:
+        step = self.responses[self.calls]
+        self.calls += 1
+        if isinstance(step, Exception):
+            raise step
+        return step
+
+
+def sync_client(counter, retry=None, monkeypatch=None):
+    if monkeypatch is not None:
+        monkeypatch.setattr("time.sleep", lambda _s: None)
+    return Stream(
+        api_key="key",
+        api_secret="secret",
+        transport=httpx.MockTransport(counter),
+        retry=retry,
+    )
+
+
+def async_client(counter, retry=None, monkeypatch=None):
+    if monkeypatch is not None:
+
+        async def no_sleep(_s):
+            return None
+
+        monkeypatch.setattr("asyncio.sleep", no_sleep)
+    return AsyncStream(
+        api_key="key",
+        api_secret="secret",
+        transport=httpx.MockTransport(counter),
+        retry=retry,
+    )
+
+
+ENABLED = RetryConfig(enabled=True, max_attempts=3, max_backoff=0.001)
+
+
+# ── sync ────────────────────────────────────────────────────────────
+
+
+def test_disabled_by_default_single_attempt(monkeypatch, caplog):
+    import logging
+
+    caplog.set_level(logging.DEBUG, logger="getstream")
+    counter = Counter(
+        [httpx.Response(429, headers={"Retry-After": "1"}, json=rate_limited_body())]
+    )
+    client = sync_client(counter, monkeypatch=monkeypatch)
+    with pytest.raises(StreamRateLimitException):
+        client.get("/api/v2/app")
+    assert counter.calls == 1
+    # A single, non-retried 429 is logged via http.response.received only;
+    # http.request.failed is never emitted for it.
+    failed = [r for r in caplog.records if r.getMessage() == "http.request.failed"]
+    assert len(failed) == 0
+
+
+def test_enabled_get_retries_429_then_succeeds(monkeypatch, caplog):
+    import logging
+
+    logger = logging.getLogger("test.retry.sync")
+    caplog.set_level(logging.DEBUG, logger="test.retry.sync")
+    counter = Counter(
+        [
+            httpx.Response(429, json=rate_limited_body()),
+            httpx.Response(200, json={}),
+        ]
+    )
+    client = Stream(
+        api_key="key",
+        api_secret="secret",
+        transport=httpx.MockTransport(counter),
+        retry=ENABLED,
+        logger=logger,
+    )
+    monkeypatch.setattr("time.sleep", lambda _s: None)
+    client.get("/api/v2/app")
+    assert counter.calls == 2
+    failed = [r for r in caplog.records if r.getMessage() == "http.request.failed"]
+    assert len(failed) == 1
+    assert failed[0].levelno == logging.DEBUG
+    # Cross-SDK rule: a retried 429 must not carry error.type (transport-only enum).
+    assert not hasattr(failed[0], "error.type")
+    assert getattr(failed[0], "retry.attempt") == 1
+
+
+def test_enabled_post_never_retried(monkeypatch):
+    counter = Counter([httpx.Response(429, json=rate_limited_body())])
+    client = sync_client(counter, retry=ENABLED, monkeypatch=monkeypatch)
+    with pytest.raises(StreamRateLimitException):
+        client.post("/api/v2/x", json={})
+    assert counter.calls == 1
+
+
+def test_unrecoverable_never_retried(monkeypatch, caplog):
+    import logging
+
+    caplog.set_level(logging.DEBUG, logger="getstream")
+    counter = Counter([httpx.Response(429, json=rate_limited_body(unrecoverable=True))])
+    client = sync_client(counter, retry=ENABLED, monkeypatch=monkeypatch)
+    with pytest.raises(StreamRateLimitException):
+        client.get("/api/v2/app")
+    assert counter.calls == 1
+    # unrecoverable 429 is never retried, so http.request.failed never fires.
+    failed = [r for r in caplog.records if r.getMessage() == "http.request.failed"]
+    assert len(failed) == 0
+
+
+def test_transport_error_retried(monkeypatch, caplog):
+    import logging
+
+    logger = logging.getLogger("test.retry.sync.transport")
+    caplog.set_level(logging.DEBUG, logger="test.retry.sync.transport")
+    counter = Counter(
+        [
+            httpx.ConnectError("reset"),
+            httpx.Response(200, json={}),
+        ]
+    )
+    client = Stream(
+        api_key="key",
+        api_secret="secret",
+        transport=httpx.MockTransport(counter),
+        retry=ENABLED,
+        logger=logger,
+    )
+    monkeypatch.setattr("time.sleep", lambda _s: None)
+    client.get("/api/v2/app")
+    assert counter.calls == 2
+    failed = [r for r in caplog.records if r.getMessage() == "http.request.failed"]
+    assert len(failed) == 1
+    assert failed[0].levelno == logging.DEBUG
+    # Transport failures DO carry error.type (the closed transport-only enum).
+    assert hasattr(failed[0], "error.type")
+
+
+def test_exhaustion_surfaces_last_error(monkeypatch, caplog):
+    import logging
+
+    caplog.set_level(logging.DEBUG, logger="getstream")
+    counter = Counter([httpx.Response(429, json=rate_limited_body())] * 3)
+    client = sync_client(counter, retry=ENABLED, monkeypatch=monkeypatch)
+    with pytest.raises(StreamRateLimitException):
+        client.get("/api/v2/app")
+    assert counter.calls == 3
+    # The 2 retried attempts log at DEBUG; the terminal (exhausting) 429 is
+    # never logged as http.request.failed at all (unlike transport
+    # exhaustion, see test_transport_exhaustion_logs_final_error) since its
+    # response was already captured via http.response.received.
+    failed = [r for r in caplog.records if r.getMessage() == "http.request.failed"]
+    assert len(failed) == 2
+    assert not any(r.levelno == logging.ERROR for r in failed)
+
+
+def test_transport_exhaustion_logs_final_error(monkeypatch, caplog):
+    import logging
+
+    logger = logging.getLogger("test.retry.sync.exhaust")
+    caplog.set_level(logging.DEBUG, logger="test.retry.sync.exhaust")
+    counter = Counter([httpx.ConnectError("reset")] * 3)
+    client = Stream(
+        api_key="key",
+        api_secret="secret",
+        transport=httpx.MockTransport(counter),
+        retry=ENABLED,
+        logger=logger,
+    )
+    monkeypatch.setattr("time.sleep", lambda _s: None)
+    with pytest.raises(StreamTransportException):
+        client.get("/api/v2/app")
+    assert counter.calls == 3
+    failed = [r for r in caplog.records if r.getMessage() == "http.request.failed"]
+    # 2 retried attempts at DEBUG, 1 final at ERROR.
+    assert len(failed) == 3
+    assert [r.levelno for r in failed] == [logging.DEBUG, logging.DEBUG, logging.ERROR]
+    assert all(hasattr(r, "error.type") for r in failed)
+
+
+def test_disabled_transport_failure_logs_identical_to_today(monkeypatch, caplog):
+    """With retry disabled (default), logging must be unchanged: exactly one
+    ERROR http.request.failed, no DEBUG retry-log noise."""
+    import logging
+
+    logger = logging.getLogger("test.retry.sync.disabled")
+    caplog.set_level(logging.DEBUG, logger="test.retry.sync.disabled")
+    counter = Counter([httpx.ConnectError("reset")])
+    client = Stream(
+        api_key="key",
+        api_secret="secret",
+        transport=httpx.MockTransport(counter),
+        logger=logger,
+    )
+    with pytest.raises(StreamTransportException):
+        client.get("/api/v2/app")
+    failed = [r for r in caplog.records if r.getMessage() == "http.request.failed"]
+    assert len(failed) == 1
+    assert failed[0].levelno == logging.ERROR
+    # Field content must match the pre-retry per-attempt failed log exactly
+    # (this is the logging-fidelity regression the retry loop must not
+    # reintroduce): method, path, non-empty endpoint name, error type/message,
+    # and duration are all present.
+    record = failed[0]
+    assert record.getMessage() == "http.request.failed"
+    assert getattr(record, "http.request.method") == "GET"
+    assert getattr(record, "url.path") == "/api/v2/app"
+    assert getattr(record, "stream.endpoint_name")
+    assert getattr(record, "error.type")
+    assert getattr(record, "error.message")
+    assert isinstance(getattr(record, "duration_ms"), int)
+
+
+def test_retry_after_honored(monkeypatch):
+    """retry_after is honored end-to-end via client.get, and clamped to
+    max_backoff (sync mirror of test_async_retry_after_honored)."""
+    slept = []
+    monkeypatch.setattr("time.sleep", slept.append)
+    counter = Counter(
+        [
+            httpx.Response(429, headers={"Retry-After": "5"}, json=rate_limited_body()),
+            httpx.Response(200, json={}),
+        ]
+    )
+    retry = RetryConfig(enabled=True, max_attempts=3, max_backoff=1.0)
+    client = Stream(
+        api_key="key",
+        api_secret="secret",
+        transport=httpx.MockTransport(counter),
+        retry=retry,
+    )
+    client.get("/api/v2/app")
+    assert counter.calls == 2
+    assert slept == [1.0]  # clamped from 5s to max_backoff=1.0
+
+
+def test_delay_clamp_and_jitter_bounds():
+    retry = RetryConfig(enabled=True, max_attempts=3, max_backoff=30.0)
+    exc = StreamRateLimitException(status_code=429, retry_after=timedelta(seconds=600))
+    assert _retry_delay(retry, exc, 0) == 30.0
+    transport = StreamTransportException("timeout")
+    for attempt in range(3):
+        ceil = min(30.0, 2.0**attempt)
+        for _ in range(50):
+            assert 0 <= _retry_delay(retry, transport, attempt) <= ceil
+
+
+def test_eligibility_matrix():
+    retry = RetryConfig(enabled=True, max_attempts=3)
+    transport = StreamTransportException("timeout")
+    assert _retry_eligible(retry, transport, "GET", 0)
+    assert _retry_eligible(retry, transport, "HEAD", 1)
+    assert not _retry_eligible(retry, transport, "POST", 0)
+    assert not _retry_eligible(retry, transport, "GET", 2)
+    assert not _retry_eligible(None, transport, "GET", 0)
+    assert not _retry_eligible(RetryConfig(), transport, "GET", 0)
+
+
+def test_unrecoverable_not_eligible():
+    retry = RetryConfig(enabled=True, max_attempts=3)
+    rate_limited = StreamRateLimitException(status_code=429, unrecoverable=True)
+    assert not _retry_eligible(retry, rate_limited, "GET", 0)
+
+
+def test_retry_config_validation():
+    with pytest.raises(ValueError):
+        RetryConfig(max_attempts=0)
+    with pytest.raises(ValueError):
+        RetryConfig(max_backoff=-1.0)
+
+
+# ── async ───────────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_async_disabled_by_default_single_attempt(caplog):
+    import logging
+
+    caplog.set_level(logging.DEBUG, logger="getstream")
+    counter = Counter([httpx.Response(429, json=rate_limited_body())])
+    client = async_client(counter)
+    with pytest.raises(StreamRateLimitException):
+        await client.get("/api/v2/app")
+    assert counter.calls == 1
+    # A single, non-retried 429 is never logged as http.request.failed.
+    failed = [r for r in caplog.records if r.getMessage() == "http.request.failed"]
+    assert len(failed) == 0
+    await client.aclose()
+
+
+@pytest.mark.asyncio
+async def test_async_enabled_get_retries(monkeypatch, caplog):
+    import logging
+
+    logger = logging.getLogger("test.retry.async.enabled")
+    caplog.set_level(logging.DEBUG, logger="test.retry.async.enabled")
+    counter = Counter(
+        [
+            httpx.Response(429, json=rate_limited_body()),
+            httpx.Response(200, json={}),
+        ]
+    )
+    client = AsyncStream(
+        api_key="key",
+        api_secret="secret",
+        transport=httpx.MockTransport(counter),
+        retry=ENABLED,
+        logger=logger,
+    )
+
+    async def no_sleep(_s):
+        return None
+
+    monkeypatch.setattr("asyncio.sleep", no_sleep)
+    await client.get("/api/v2/app")
+    assert counter.calls == 2
+    failed = [r for r in caplog.records if r.getMessage() == "http.request.failed"]
+    assert len(failed) == 1
+    assert failed[0].levelno == logging.DEBUG
+    # Cross-SDK rule: a retried 429 must not carry error.type (transport-only enum).
+    assert not hasattr(failed[0], "error.type")
+    await client.aclose()
+
+
+@pytest.mark.asyncio
+async def test_async_post_never_retried(monkeypatch):
+    counter = Counter([httpx.Response(429, json=rate_limited_body())])
+    client = async_client(counter, retry=ENABLED, monkeypatch=monkeypatch)
+    with pytest.raises(StreamRateLimitException):
+        await client.post("/api/v2/x", json={})
+    assert counter.calls == 1
+    await client.aclose()
+
+
+@pytest.mark.asyncio
+async def test_async_unrecoverable_never_retried(monkeypatch, caplog):
+    import logging
+
+    caplog.set_level(logging.DEBUG, logger="getstream")
+    counter = Counter([httpx.Response(429, json=rate_limited_body(unrecoverable=True))])
+    client = async_client(counter, retry=ENABLED, monkeypatch=monkeypatch)
+    with pytest.raises(StreamRateLimitException):
+        await client.get("/api/v2/app")
+    assert counter.calls == 1
+    # unrecoverable 429 is never retried, so http.request.failed never fires.
+    failed = [r for r in caplog.records if r.getMessage() == "http.request.failed"]
+    assert len(failed) == 0
+    await client.aclose()
+
+
+@pytest.mark.asyncio
+async def test_async_transport_error_retried(monkeypatch, caplog):
+    import logging
+
+    logger = logging.getLogger("test.retry.async.transport")
+    caplog.set_level(logging.DEBUG, logger="test.retry.async.transport")
+    counter = Counter(
+        [
+            httpx.ConnectError("reset"),
+            httpx.Response(200, json={}),
+        ]
+    )
+    client = AsyncStream(
+        api_key="key",
+        api_secret="secret",
+        transport=httpx.MockTransport(counter),
+        retry=ENABLED,
+        logger=logger,
+    )
+
+    async def no_sleep(_s):
+        return None
+
+    monkeypatch.setattr("asyncio.sleep", no_sleep)
+    await client.get("/api/v2/app")
+    assert counter.calls == 2
+    failed = [r for r in caplog.records if r.getMessage() == "http.request.failed"]
+    assert len(failed) == 1
+    assert failed[0].levelno == logging.DEBUG
+    # Transport failures DO carry error.type (the closed transport-only enum).
+    assert hasattr(failed[0], "error.type")
+    assert getattr(failed[0], "error.type") != "rate_limited"
+    await client.aclose()
+
+
+@pytest.mark.asyncio
+async def test_async_exhaustion_surfaces_last_error(monkeypatch, caplog):
+    import logging
+
+    caplog.set_level(logging.DEBUG, logger="getstream")
+    counter = Counter([httpx.Response(429, json=rate_limited_body())] * 3)
+    client = async_client(counter, retry=ENABLED, monkeypatch=monkeypatch)
+    with pytest.raises(StreamRateLimitException):
+        await client.get("/api/v2/app")
+    assert counter.calls == 3
+    # The 2 retried attempts log at DEBUG; the terminal (exhausting) 429 is
+    # never logged as http.request.failed (its response was already
+    # captured via http.response.received).
+    failed = [r for r in caplog.records if r.getMessage() == "http.request.failed"]
+    assert len(failed) == 2
+    assert not any(r.levelno == logging.ERROR for r in failed)
+    await client.aclose()
+
+
+@pytest.mark.asyncio
+async def test_async_retry_after_honored(monkeypatch):
+    """retry_after honored on the async path too, and clamped to max_backoff."""
+    slept = []
+
+    async def capture_sleep(seconds):
+        slept.append(seconds)
+
+    monkeypatch.setattr("asyncio.sleep", capture_sleep)
+    counter = Counter(
+        [
+            httpx.Response(429, headers={"Retry-After": "5"}, json=rate_limited_body()),
+            httpx.Response(200, json={}),
+        ]
+    )
+    retry = RetryConfig(enabled=True, max_attempts=3, max_backoff=1.0)
+    client = AsyncStream(
+        api_key="key",
+        api_secret="secret",
+        transport=httpx.MockTransport(counter),
+        retry=retry,
+    )
+    await client.get("/api/v2/app")
+    assert counter.calls == 2
+    assert slept == [1.0]  # clamped from 5s to max_backoff=1.0
+    await client.aclose()