diff --git a/CHANGELOG.md b/CHANGELOG.md
index 04d29596..413302bf 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -45,7 +45,27 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
     - `request_timeout: float` (seconds): default `30.0` (was `6.0`; see Behavior changes)
 
   These tune the underlying `httpx.Limits` and `httpx.Timeout`. The existing `http_client=` and `transport=` kwargs continue to act as escape hatches; when `http_client` is set, none of the four new kwargs apply. Env-var fallbacks for the new kwargs: `STREAM_MAX_CONNS_PER_HOST`, `STREAM_IDLE_TIMEOUT`, `STREAM_CONNECT_TIMEOUT`, `STREAM_REQUEST_TIMEOUT`.
-- INFO log on client construction (logger `getstream`) lists the effective pool config and whether a user-supplied `http_client` is in use.
+
+- Structured logging ([CHA-2957](https://linear.app/stream/issue/CHA-2957)).
+  New `logger: logging.Logger | None` and `log_bodies: bool = False` kwargs on
+  `Stream(...)` and `AsyncStream(...)`. Off by default: nothing is emitted
+  unless a logger is passed. Four events, each carrying structured fields via
+  the stdlib logging `extra={}` mechanism:
+    - `client.initialized` (INFO, once at construction): SDK name/version,
+      resolved pool knobs, `gzip_enabled`, `user_http_client`, `log_bodies`.
+      Replaces the old plain-text pool-config INFO line.
+    - `http.request.sent` (DEBUG, before each request): method, path,
+      redacted query, `stream.endpoint_name`.
+    - `http.response.received` (DEBUG, after any response including
+      4xx/5xx): status code, response size, `duration_ms`. HTTP error
+      status codes are data, not failures.
+    - `http.request.failed` (ERROR, transport failure only, no HTTP
+      response received): `error.type` (`connection_reset` / `timeout` /
+      `dns_failure` / `tls_handshake_failed` / `unknown`), `error.message`.
+  Query values for `api_key`/`api_secret`/`token` and top-level JSON body
+  keys `api_secret`/`token`/`password` are always redacted; no opt-out.
+  Bodies are never logged by default; `log_bodies=True` adds redacted
+  request/response bodies and emits one WARNING at construction.
 
 - Webhook handling spec helpers (CHA-2961): `UnknownEvent` dataclass for
   forward-compat; `gunzip_payload`, `decode_sqs_payload`, `decode_sns_payload`
diff --git a/README.md b/README.md
index 705512fd..f0327260 100644
--- a/README.md
+++ b/README.md
@@ -99,6 +99,19 @@ response: StreamResponse[StartClosedCaptionsResponse] = call.start_closed_captio
 response.data  # Gives the StartClosedCaptionsResponse model
 ```
 
+### Logging
+
+The SDK emits structured log events (`client.initialized`, `http.request.sent`, `http.response.received`, `http.request.failed`) through the stdlib `logging` module. By default nothing is printed: pass a `logging.Logger` to see them.
+
+```python
+import logging
+
+logging.basicConfig(level=logging.DEBUG)
+client = Stream(api_key="key", api_secret="secret", logger=logging.getLogger("myapp.stream"))
+```
+
+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).
+
 ### App configuration
 
 ```python
diff --git a/getstream/__init__.py b/getstream/__init__.py
index 6e6dea52..066a59a7 100644
--- a/getstream/__init__.py
+++ b/getstream/__init__.py
@@ -1,3 +1,5 @@
+import logging
+
 from getstream.exceptions import (  # noqa: F401
     StreamApiException,
     StreamException,
@@ -7,3 +9,7 @@
 )
 from getstream.stream import Stream  # noqa: F401
 from getstream.stream import AsyncStream  # noqa: F401
+
+# No-op until the caller attaches a handler: the SDK never configures the
+# logger's level or output, it only emits at each event's documented level.
+logging.getLogger("getstream").addHandler(logging.NullHandler())
diff --git a/getstream/base.py b/getstream/base.py
index be4a48cf..5e9c5bc9 100644
--- a/getstream/base.py
+++ b/getstream/base.py
@@ -13,10 +13,12 @@
     build_api_exception,
     wrap_transport_error,
 )
+from getstream.logging_utils import redact_json_body, redact_query
 from getstream.stream_response import StreamResponse
 from getstream.generic import T
 import httpx
 from getstream.config import BaseConfig
+from getstream.version import VERSION
 from urllib.parse import quote
 from abc import ABC
 from getstream.common.telemetry import (
@@ -59,22 +61,46 @@ def _resolve_pool_knobs(obj):
     )
 
 
-def _log_pool_config(cfg, *, user_http_client: bool) -> None:
-    if user_http_client:
-        logger.info(
-            "getstream connection pool: user_http_client=True (5 knobs not applied)"
-        )
-    else:
-        logger.info(
-            "getstream connection pool: "
-            "max_conns_per_host=%s idle_timeout=%ss "
-            "connect_timeout=%ss request_timeout=%ss "
-            "user_http_client=False",
-            cfg.max_conns_per_host,
-            cfg.idle_timeout,
-            cfg.connect_timeout,
-            cfg.timeout,
-        )
+def _resolve_logger(obj) -> logging.Logger:
+    """The caller's injected logger (``Stream``/``AsyncStream``'s ``logger=``
+    kwarg, plumbed onto ``obj.log`` the same way as the pool knobs), or the
+    shared module logger when none was passed."""
+    return getattr(obj, "log", None) or logger
+
+
+def _log_client_initialized(cfg, *, user_http_client: bool) -> None:
+    """Emit the one-shot ``client.initialized`` event, replacing the old
+    plain-text pool-config INFO line with the structured logging schema."""
+    _resolve_logger(cfg).info(
+        "client.initialized",
+        extra={
+            "stream.sdk.name": "stream-py",
+            "stream.sdk.version": VERSION,
+            "stream.client.max_conns_per_host": cfg.max_conns_per_host,
+            "stream.client.idle_timeout_seconds": cfg.idle_timeout,
+            "stream.client.connect_timeout_seconds": cfg.connect_timeout,
+            "stream.client.request_timeout_seconds": cfg.timeout,
+            "stream.client.gzip_enabled": not user_http_client,
+            "stream.client.user_http_client": user_http_client,
+            "stream.client.log_bodies": bool(getattr(cfg, "log_bodies", False)),
+        },
+    )
+
+
+def _response_body_for_log(response: httpx.Response):
+    """Redact a response body for the ``http.response.body`` log field.
+    JSON bodies get the shallow key redaction; anything else (or anything
+    that fails to parse as JSON) is passed through as text."""
+    # Media types are case-insensitive; match application/json and any
+    # structured +json type (e.g. application/problem+json) so their bodies
+    # are redacted, not logged as raw text.
+    content_type = response.headers.get("content-type", "").lower()
+    if "application/json" in content_type or "+json" in content_type:
+        try:
+            return redact_json_body(json.loads(response.text))
+        except (ValueError, TypeError):
+            return response.text
+    return response.text
 
 
 def _read_file_bytes(file_path: str) -> bytes:
@@ -284,6 +310,18 @@ def _request_sync(
         url_path, url_full, endpoint, attrs = self._prepare_request(
             method, path, query_params, kwargs
         )
+        log = _resolve_logger(self)
+        log_bodies = bool(getattr(self, "log_bodies", False))
+        sent_extra = {
+            "http.request.method": method,
+            "url.path": path,
+            "url.query": redact_query(query_params) or "",
+            "stream.endpoint_name": endpoint,
+        }
+        if log_bodies:
+            sent_extra["http.request.body"] = redact_json_body(kwargs.get("json"))
+        log.debug("http.request.sent", extra=sent_extra)
+
         start = time.perf_counter()
         # Span name uses logical operation (endpoint) rather than raw HTTP
         with span_request(
@@ -296,7 +334,19 @@ def _request_sync(
                     url_path, params=query_params, *args, **call_kwargs
                 )
             except httpx.RequestError as err:
-                raise wrap_transport_error(err) from 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
             duration = parse_duration_from_body(response.content)
             if duration:
                 span.set_attribute("http.server.duration", duration)
@@ -308,6 +358,17 @@ def _request_sync(
                 pass
 
             duration_ms = (time.perf_counter() - start) * 1000.0
+            received_extra = {
+                "http.request.method": method,
+                "url.path": path,
+                "stream.endpoint_name": endpoint,
+                "http.response.status_code": response.status_code,
+                "http.response.body.size": len(response.content or b""),
+                "duration_ms": int(duration_ms),
+            }
+            if log_bodies:
+                received_extra["http.response.body"] = _response_body_for_log(response)
+            log.debug("http.response.received", extra=received_extra)
             # Metrics should be low-cardinality: exclude url/call_cid/channel_cid
             metric_attrs = metric_attributes(
                 api_key=self.api_key,
@@ -593,6 +654,18 @@ async def _request_async(
         url_path, url_full, endpoint, attrs = self._prepare_request(
             method, path, query_params, kwargs
         )
+        log = _resolve_logger(self)
+        log_bodies = bool(getattr(self, "log_bodies", False))
+        sent_extra = {
+            "http.request.method": method,
+            "url.path": path,
+            "url.query": redact_query(query_params) or "",
+            "stream.endpoint_name": endpoint,
+        }
+        if log_bodies:
+            sent_extra["http.request.body"] = redact_json_body(kwargs.get("json"))
+        log.debug("http.request.sent", extra=sent_extra)
+
         start = time.perf_counter()
         with span_request(
             endpoint, attributes=attrs, request_body=kwargs.get("json")
@@ -612,7 +685,19 @@ async def _request_async(
                     url_path, params=query_params, *args, **call_kwargs
                 )
             except httpx.RequestError as err:
-                raise wrap_transport_error(err) from 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
             duration = parse_duration_from_body(response.content)
             if duration:
                 span.set_attribute("http.server.duration", duration)
@@ -624,6 +709,19 @@ async def _request_async(
                 pass
 
             duration_ms = (time.perf_counter() - start) * 1000.0
+            received_extra = {
+                "http.request.method": method,
+                "url.path": path,
+                "stream.endpoint_name": endpoint,
+                "http.response.status_code": response.status_code,
+                "http.response.body.size": len(response.content or b""),
+                "duration_ms": int(duration_ms),
+            }
+            if log_bodies:
+                received_extra["http.response.body"] = await asyncio.to_thread(
+                    _response_body_for_log, response
+                )
+            log.debug("http.response.received", extra=received_extra)
             # Metrics should be low-cardinality: exclude url/call_cid/channel_cid
             metric_attrs = metric_attributes(
                 api_key=self.api_key,
diff --git a/getstream/logging_utils.py b/getstream/logging_utils.py
new file mode 100644
index 00000000..52487f29
--- /dev/null
+++ b/getstream/logging_utils.py
@@ -0,0 +1,21 @@
+"""Redaction helpers for the SDK's structured log events. Secret values are
+replaced with a literal marker; the scan is shallow by design."""
+
+REDACTED = "<redacted>"
+REDACTED_QUERY_PARAMS = {"api_key", "api_secret", "token"}
+REDACTED_BODY_KEYS = {"api_secret", "token", "password"}
+
+
+def redact_query(params):
+    if not params:
+        return params
+    return {
+        k: (REDACTED if k.lower() in REDACTED_QUERY_PARAMS else v)
+        for k, v in dict(params).items()
+    }
+
+
+def redact_json_body(body):
+    if not isinstance(body, dict):
+        return body
+    return {k: (REDACTED if k in REDACTED_BODY_KEYS else v) for k, v in body.items()}
diff --git a/getstream/stream.py b/getstream/stream.py
index 7ff71910..eb326062 100644
--- a/getstream/stream.py
+++ b/getstream/stream.py
@@ -2,6 +2,7 @@
 
 from contextlib import AsyncExitStack
 from functools import cached_property
+import logging
 import time
 from typing import List, Optional
 from uuid import uuid4
@@ -10,7 +11,7 @@
 import jwt
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
-from getstream.base import _log_pool_config
+from getstream.base import _log_client_initialized, _resolve_logger
 from getstream.common import telemetry
 from getstream.chat.client import ChatClient
 from getstream.chat.async_client import ChatClient as AsyncChatClient
@@ -92,6 +93,8 @@ def __init__(
         max_conns_per_host: Optional[int] = None,
         idle_timeout: Optional[float] = None,
         connect_timeout: Optional[float] = None,
+        logger: Optional[logging.Logger] = None,
+        log_bodies: bool = False,
     ):
         """Build a Stream client.
 
@@ -114,6 +117,8 @@ def __init__(
             max_conns_per_host: Max concurrent TCP connections per host. Default 5. Ignored when ``http_client`` is set.
             idle_timeout: Idle connection lifetime in seconds. Default 55.0 (sits 5s under the typical 60s LB idle timeout). Ignored when ``http_client`` is set.
             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.
 
         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.
@@ -189,6 +194,13 @@ def _settings() -> _PoolSettings:
         self._transport = transport
         self._http_client = http_client
         self.token = token or self._create_token()
+        # log / log_bodies are read by BaseClient via getattr(self, ...), same
+        # plumbing as the pool knobs below: the intermediate generated REST
+        # clients (CommonRestClient etc.) do not forward these kwargs, so they
+        # must be set on self before super().__init__() and copied onto
+        # sub-clients in _apply_shared_client.
+        self.log = logger
+        self.log_bodies = log_bodies
         # 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
@@ -203,9 +215,17 @@ def _settings() -> _PoolSettings:
         # the parent's client avoids that and keeps one pool per Stream.
         self._shared_client = self.client
 
-        # Emit the pool-config INFO line exactly once per Stream, reflecting the
-        # resolved knobs on the top-level client. Sub-clients no longer log.
-        _log_pool_config(self, user_http_client=http_client is not None)
+        # Emit the client.initialized event exactly once per Stream, reflecting
+        # the resolved knobs on the top-level client. Sub-clients no longer log
+        # their own construction.
+        _log_client_initialized(self, user_http_client=http_client is not None)
+        if self.log_bodies:
+            _resolve_logger(self).warning(
+                "HTTP request/response bodies will be logged. Auth headers "
+                "and known-secret fields are still redacted, but other "
+                "sensitive data (messages, PII) may appear in logs. Disable "
+                "for production."
+            )
 
     @property
     def api_secret(self) -> str:
@@ -234,6 +254,11 @@ def _apply_shared_client(self, sub_client):
                 sub_client.client.close()
             sub_client.client = self._shared_client
             sub_client._owns_http_client = False
+        # log / log_bodies: same getattr(self, ..., default) plumbing as the
+        # pool knobs, so sub-clients (which issue the actual requests) emit
+        # 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)
         return sub_client
 
     def create_token(
@@ -284,6 +309,8 @@ def clone_for_token(self, token: str):
             idle_timeout=self.idle_timeout,
             connect_timeout=self.connect_timeout,
             user_agent=self.user_agent,
+            logger=self.log,
+            log_bodies=self.log_bodies,
         )
 
     def create_call_token(
@@ -543,6 +570,8 @@ def as_async(self) -> "AsyncStream":
             connect_timeout=self.connect_timeout,
             base_url=self.base_url,
             user_agent=self.user_agent,
+            logger=self.log,
+            log_bodies=self.log_bodies,
         )
 
     @cached_property
diff --git a/tests/test_http_client.py b/tests/test_http_client.py
index b440b4be..fbe7d8e0 100644
--- a/tests/test_http_client.py
+++ b/tests/test_http_client.py
@@ -409,7 +409,10 @@ async def test_pool_knobs_ignored_when_http_client_provided(self):
         await client.aclose()
 
 
-# ── INFO log on construction ─────────────────────────────────────────
+# ── client.initialized event on construction (CHA-2957) ──────────────
+# Formerly a plain-text "getstream connection pool: ..." INFO line
+# (_log_pool_config); replaced by the structured client.initialized event.
+# See tests/test_logging.py for the full logging-feature test suite.
 
 
 class TestSyncInfoLog:
@@ -418,12 +421,13 @@ def test_info_log_emitted_with_defaults(self, caplog):
             Stream(api_key="k", api_secret="s", base_url="http://test")
         infos = [r for r in caplog.records if r.name == "getstream"]
         assert len(infos) == 1
-        msg = infos[0].getMessage()
-        assert "max_conns_per_host=5" in msg
-        assert "idle_timeout=55.0s" in msg
-        assert "connect_timeout=10.0s" in msg
-        assert "request_timeout=30.0s" in msg
-        assert "user_http_client=False" in msg
+        r = infos[0]
+        assert r.getMessage() == "client.initialized"
+        assert getattr(r, "stream.client.max_conns_per_host") == 5
+        assert getattr(r, "stream.client.idle_timeout_seconds") == 55.0
+        assert getattr(r, "stream.client.connect_timeout_seconds") == 10.0
+        assert getattr(r, "stream.client.request_timeout_seconds") == 30.0
+        assert getattr(r, "stream.client.user_http_client") is False
 
     def test_info_log_emitted_once_even_with_sub_clients(self, caplog):
         # Regression for CHA-2956 MINOR: the pool-config INFO line must fire
@@ -448,11 +452,11 @@ def test_info_log_reflects_resolved_knobs(self, caplog):
             )
         infos = [r for r in caplog.records if r.name == "getstream"]
         assert len(infos) == 1
-        msg = infos[0].getMessage()
-        assert "max_conns_per_host=33" in msg
-        assert "idle_timeout=12.0s" in msg
-        assert "connect_timeout=4.0s" in msg
-        assert "request_timeout=9.0s" in msg
+        r = infos[0]
+        assert getattr(r, "stream.client.max_conns_per_host") == 33
+        assert getattr(r, "stream.client.idle_timeout_seconds") == 12.0
+        assert getattr(r, "stream.client.connect_timeout_seconds") == 4.0
+        assert getattr(r, "stream.client.request_timeout_seconds") == 9.0
 
     def test_info_log_when_user_http_client_provided(self, caplog):
         custom = httpx.Client(transport=_mock_transport())
@@ -465,7 +469,8 @@ def test_info_log_when_user_http_client_provided(self, caplog):
             )
         infos = [r for r in caplog.records if r.name == "getstream"]
         assert len(infos) == 1
-        assert "user_http_client=True" in infos[0].getMessage()
+        assert getattr(infos[0], "stream.client.user_http_client") is True
+        assert getattr(infos[0], "stream.client.gzip_enabled") is False
 
 
 @pytest.mark.asyncio
@@ -476,8 +481,8 @@ async def test_info_log_emitted_with_defaults(self, caplog):
             await client.aclose()
         infos = [r for r in caplog.records if r.name == "getstream"]
         assert len(infos) == 1
-        assert "max_conns_per_host=5" in infos[0].getMessage()
-        assert "user_http_client=False" in infos[0].getMessage()
+        assert getattr(infos[0], "stream.client.max_conns_per_host") == 5
+        assert getattr(infos[0], "stream.client.user_http_client") is False
 
 
 # ── sync/async parity ────────────────────────────────────────────────
diff --git a/tests/test_logging.py b/tests/test_logging.py
new file mode 100644
index 00000000..6054bbbc
--- /dev/null
+++ b/tests/test_logging.py
@@ -0,0 +1,155 @@
+import logging
+
+import httpx
+import pytest
+
+from getstream import AsyncStream, Stream
+from getstream.logging_utils import redact_json_body, redact_query
+
+
+def make_client(handler, caplog, **kwargs):
+    logger = logging.getLogger("test.stream.logging")
+    caplog.set_level(logging.DEBUG, logger="test.stream.logging")
+    return Stream(
+        api_key="key",
+        api_secret="secret",
+        transport=httpx.MockTransport(handler),
+        logger=logger,
+        **kwargs,
+    )
+
+
+def make_async_client(handler, caplog, **kwargs):
+    logger = logging.getLogger("test.stream.logging.async")
+    caplog.set_level(logging.DEBUG, logger="test.stream.logging.async")
+    return AsyncStream(
+        api_key="key",
+        api_secret="secret",
+        transport=httpx.MockTransport(handler),
+        logger=logger,
+        **kwargs,
+    )
+
+
+def ok(_request):
+    return httpx.Response(200, json={"ok": True})
+
+
+def records_named(caplog, name):
+    return [r for r in caplog.records if r.getMessage() == name]
+
+
+def test_client_initialized_once_with_schema(caplog):
+    make_client(ok, caplog)
+    inits = records_named(caplog, "client.initialized")
+    assert len(inits) == 1
+    r = inits[0]
+    assert getattr(r, "stream.sdk.name", None) == "stream-py"
+    assert hasattr(r, "stream.client.max_conns_per_host")
+    assert getattr(r, "stream.client.log_bodies") is False
+
+
+def test_sent_and_received_on_success(caplog):
+    client = make_client(ok, caplog)
+    client.get("/api/v2/app")
+    assert len(records_named(caplog, "http.request.sent")) == 1
+    received = records_named(caplog, "http.response.received")
+    assert len(received) == 1
+    assert getattr(received[0], "http.response.status_code") == 200
+    assert hasattr(received[0], "duration_ms")
+
+
+def test_error_status_is_received_not_failed(caplog):
+    client = make_client(
+        lambda _r: httpx.Response(500, json={"code": 1, "message": "boom"}), caplog
+    )
+    with pytest.raises(Exception):
+        client.get("/api/v2/app")
+    assert len(records_named(caplog, "http.response.received")) == 1
+    assert not records_named(caplog, "http.request.failed")
+
+
+def test_transport_failure_emits_failed(caplog):
+    def boom(_request):
+        raise httpx.ConnectError("reset")
+
+    client = make_client(boom, caplog)
+    with pytest.raises(Exception):
+        client.get("/api/v2/app")
+    failed = records_named(caplog, "http.request.failed")
+    assert len(failed) == 1
+    assert failed[0].levelno == logging.ERROR
+    assert getattr(failed[0], "error.type") in {
+        "connection_reset",
+        "timeout",
+        "dns_failure",
+        "tls_handshake_failed",
+        "unknown",
+    }
+
+
+def test_no_logger_no_output(caplog):
+    caplog.set_level(logging.DEBUG)
+    client = Stream(
+        api_key="key", api_secret="secret", transport=httpx.MockTransport(ok)
+    )
+    client.get("/api/v2/app")
+    assert not [r for r in caplog.records if r.name.startswith("test.")]
+
+
+def test_query_redaction(caplog):
+    client = make_client(ok, caplog)
+    client.get("/api/v2/app", query_params={"api_key": "sekret", "user_id": "123"})
+    sent = records_named(caplog, "http.request.sent")[0]
+    q = str(getattr(sent, "url.query", ""))
+    # Secret value redacted, non-secret param retained.
+    assert "sekret" not in q
+    assert "<redacted>" in q
+    assert "123" in q
+
+
+def test_log_bodies_opt_in_and_warn(caplog):
+    client = make_client(ok, caplog, log_bodies=True)
+    warns = [
+        r
+        for r in caplog.records
+        if r.levelno == logging.WARNING and "bodies will be logged" in r.getMessage()
+    ]
+    assert len(warns) == 1
+    # Secret value deliberately does not share a substring with its own key
+    # name ("token"), otherwise "tok" (the value) would trivially still be
+    # found in "token" (the key) even after correct redaction.
+    client.post("/api/v2/x", json={"token": "shh12345", "keep": "v"})
+    sent = records_named(caplog, "http.request.sent")[0]
+    body = getattr(sent, "http.request.body", "")
+    assert "shh12345" not in str(body)
+    assert "keep" in str(body)
+
+
+def test_redaction_helpers():
+    assert redact_query({"api_key": "k", "q": "x"}) == {
+        "api_key": "<redacted>",
+        "q": "x",
+    }
+    out = redact_json_body(
+        {"token": "t", "password": "p", "api_secret": "s", "keep": "v"}
+    )
+    assert out == {
+        "token": "<redacted>",
+        "password": "<redacted>",
+        "api_secret": "<redacted>",
+        "keep": "v",
+    }
+    assert redact_json_body("not json") == "not json"
+
+
+@pytest.mark.asyncio
+async def test_async_sent_and_received_on_success(caplog):
+    client = make_async_client(ok, caplog)
+    await client.get("/api/v2/app")
+    assert len(records_named(caplog, "http.request.sent")) == 1
+    received = records_named(caplog, "http.response.received")
+    assert len(received) == 1
+    assert getattr(received[0], "http.response.status_code") == 200
+    assert hasattr(received[0], "duration_ms")
+    await client.aclose()