diff --git a/docs/implementation-plan.md b/docs/implementation-plan.md
index 12be2c7a..e83b1980 100644
--- a/docs/implementation-plan.md
+++ b/docs/implementation-plan.md
@@ -375,8 +375,8 @@ defaults (per Square: `FAIL_ON_UNKNOWN_PROPERTIES=false`, `WRITE_DATES_AS_TIMEST
 > Superseded by #30 (pagination unification): `Page` now exposes the raw per-page `Response` and is `Closeable` (materialized `items` and derived `statusCode` / `headers` / `request` survive `close()`), strategies return `PageInfo` (`nextRequest == null` = end of stream), and `SimplePage` was removed.
 
 **Status: shipped.** `Page`, `Paginator`, `PaginationStrategy`, and the three strategies
-(`Cursor` / `PageNumber` / `LinkHeader`) are in `sdk-core/.../pagination`, alongside
-helper types `SimplePage` and `RequestRebuilder`. `Paginator` gained a `maxPages` safety cap
+(`Cursor` / `PageNumber` / `LinkHeader`) are in `sdk-core/.../pagination`, alongside the
+helper type `RequestRebuilder`. `Paginator` gained a `maxPages` safety cap
 (default `Long.MAX_VALUE`) beyond the original sketch, to bound runaway iteration against servers
 that never advance their cursor.
 
diff --git a/docs/refs-comparison.md b/docs/refs-comparison.md
index 4023466b..6622722e 100644
--- a/docs/refs-comparison.md
+++ b/docs/refs-comparison.md
@@ -230,7 +230,7 @@ below records where each scheme's design was sourced from:
 ### Pagination
 
 `Paginator<T>` + `Page<T>` ship in `sdk-core`, driven by a `PaginationStrategy` (cursor,
-page-number, token, link-header), and `pagination.PagedIterable` wraps the result. Reference
+page-number, link-header), and `pagination.PagedIterable` wraps the result. Reference
 designs we drew on:
 
 - **Square**: `SyncPagingIterable<T>` (`Iterable<T>` lazy iterator), `SyncPage<T>` (per-page holder), `BiDirectionalPage<T>` (forward + backward cursors), `CustomPager<T>` (user-implementation stub for HATEOAS).
@@ -240,7 +240,7 @@ designs we drew on:
 **What shipped, and what's left:**
 
 1. `Paginator<T>` and `Page<T>` are in `sdk-core`. `iterateAll()` returns a lazy `Iterable<T>`; `streamAll()` returns a Java 8 `Stream<T>`. Each call hands back an independent iterator with its own state.
-2. The strategy is injected via `PaginationStrategy`, with concrete impls covering cursor (`next_cursor` / `prev_cursor`), page-number, token, and link-header (RFC 8288). A `maxPages` cap guards against servers that never advance their cursor.
+2. The strategy is injected via `PaginationStrategy`, with three concrete impls: cursor (`CursorPaginationStrategy`, forward-only), page-number, and link-header (RFC 8288). Token-style APIs (`next_page_token`, `pageToken`, …) reuse `CursorPaginationStrategy` with a configurable query-param name, so no separate token strategy ships (see `architecture.md`). A `maxPages` cap guards against servers that never advance their cursor.
 3. Async variants for `sdk-async-coroutines` (`Flow<T>`) and `sdk-async-reactor` (`Flux<T>`) are not yet built.
 4. `BiDirectionalPage` is deferred until a real API needs it; Square's pattern is good when needed.
 
@@ -333,7 +333,7 @@ adapters where noted):
 - **Idempotency-key step.** Auto-injects `Idempotency-Key: UUID.randomUUID()` for `POST`/`PUT`/`PATCH`; caller-set header wins; pluggable key strategy. [`pipeline/step/IdempotencyKeyStep.kt`]
 - **Auth.** `Credential` family + RFC 7235 challenge parsing + Basic/Digest/Composite `ChallengeHandler`s + `AuthStep` pillar. [`auth/`, `http/pipeline/steps/`]
 - **`sdk-serde-jackson` adapter.** Kotlin + JSR-310 + Jdk8 modules; `FAIL_ON_UNKNOWN_PROPERTIES` and `WRITE_DATES_AS_TIMESTAMPS` disabled; `Tristate<T>` via `TristateModule`.
-- **Pagination primitives.** `Paginator<T>` + `Page<T>` + `PaginationStrategy` (cursor / page-number / token / link-header) with a `maxPages` cap; `PagedIterable` wrapper.
+- **Pagination primitives.** `Paginator<T>` + `Page<T>` + `PaginationStrategy` (cursor / page-number / link-header) with a `maxPages` cap; `PagedIterable` wrapper.
 - **Client identity header.** `ClientIdentityStep` building the `dexpace-sdk/<ver> jvm/<javaver>` token line.
 - **Tracer event vocabulary + metrics seam.** `HttpTracer` with named retry/request/response events; `Meter`/`LongCounter`/`DoubleHistogram` separate from tracing.
 - **SSE streaming.** WHATWG reader in `sdk-core`; backpressured `Flux<ServerSentEvent>` in `sdk-async-reactor`.
diff --git a/sdk-core/api/sdk-core.api b/sdk-core/api/sdk-core.api
index c27b252c..49f62f44 100644
--- a/sdk-core/api/sdk-core.api
+++ b/sdk-core/api/sdk-core.api
@@ -406,6 +406,7 @@ public final class org/dexpace/sdk/core/http/common/Headers$Builder {
 	public final fun add (Lorg/dexpace/sdk/core/http/common/HttpHeaderName;Ljava/lang/String;)Lorg/dexpace/sdk/core/http/common/Headers$Builder;
 	public final fun add (Lorg/dexpace/sdk/core/http/common/HttpHeaderName;Ljava/util/List;)Lorg/dexpace/sdk/core/http/common/Headers$Builder;
 	public final fun addAll (Lorg/dexpace/sdk/core/http/common/Headers;)Lorg/dexpace/sdk/core/http/common/Headers$Builder;
+	public final fun addUnsafeNonAscii (Ljava/lang/String;Ljava/lang/String;)Lorg/dexpace/sdk/core/http/common/Headers$Builder;
 	public final fun build ()Lorg/dexpace/sdk/core/http/common/Headers;
 	public final fun remove (Ljava/lang/String;)Lorg/dexpace/sdk/core/http/common/Headers$Builder;
 	public final fun remove (Lorg/dexpace/sdk/core/http/common/HttpHeaderName;)Lorg/dexpace/sdk/core/http/common/Headers$Builder;
@@ -1831,6 +1832,15 @@ public final class org/dexpace/sdk/core/instrumentation/ClientLogger {
 public final class org/dexpace/sdk/core/instrumentation/ClientLogger$Companion {
 }
 
+public final class org/dexpace/sdk/core/instrumentation/DroppedHeaderLogging : java/lang/Enum {
+	public static final field ONCE_PER_HEADER Lorg/dexpace/sdk/core/instrumentation/DroppedHeaderLogging;
+	public static final field PER_OCCURRENCE Lorg/dexpace/sdk/core/instrumentation/DroppedHeaderLogging;
+	public static final field VERBOSE_ONLY Lorg/dexpace/sdk/core/instrumentation/DroppedHeaderLogging;
+	public static fun getEntries ()Lkotlin/enums/EnumEntries;
+	public static fun valueOf (Ljava/lang/String;)Lorg/dexpace/sdk/core/instrumentation/DroppedHeaderLogging;
+	public static fun values ()[Lorg/dexpace/sdk/core/instrumentation/DroppedHeaderLogging;
+}
+
 public abstract interface class org/dexpace/sdk/core/instrumentation/HttpTracer {
 	public fun attemptFailed (Ljava/lang/Throwable;Ljava/lang/Long;)V
 	public fun attemptRetriesExhausted (Ljava/lang/Throwable;)V
diff --git a/sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/common/HeaderValidation.kt b/sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/common/HeaderValidation.kt
index dad0c4ae..6dd90ee8 100644
--- a/sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/common/HeaderValidation.kt
+++ b/sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/common/HeaderValidation.kt
@@ -28,19 +28,22 @@ package org.dexpace.sdk.core.http.common
  *    and `0x7F`), which covers CR, LF, and NUL. An embedded `\r`/`\n` is the same
  *    request/header-splitting vector guarded against for header values: once the name is
  *    serialised an attacker could inject a new header or a second request. A NUL or other control
- *    character is illegal in a field-name, and the two reference transports handle it differently at
- *    their raw API (OkHttp's `addHeader` throws unchecked, the JDK builder drops it); their adapters
- *    now catch and drop uniformly, but a splitting vector should never get that far. Validating here
- *    rejects it loudly at construction — fast, uniform, and transport-independent.
+ *    character is illegal in a field-name; validating here rejects it loudly at construction —
+ *    fast, uniform, and transport-independent.
+ *  - **Any non-ASCII byte** (code point `>= 0x80`). RFC 7230 field-names are ASCII `token`s, and
+ *    every shipped reference transport (OkHttp, the JDK `HttpClient`) rejects a non-ASCII name at
+ *    its own model layer, so a name no transport can put on the wire is refused here at
+ *    construction rather than accepted by the model and then silently dropped mid-dispatch.
  *
- * Policy: the control-character set is intentionally narrower than RFC 7230's full `tchar`
- * allow-list — restricting names to `tchar` would reject some non-ASCII names that certain
- * transports accept, whereas the control-character set is illegal everywhere and covers the
- * splitting/injection surface. This mirrors the conservative stance taken for values in
+ * Policy: the accepted set is printable ASCII (`0x20`–`0x7E`) — still wider than RFC 7230's
+ * `tchar`, so a name a transport's stricter grammar rejects (e.g. an interior space) is left for
+ * the transport to drop rather than pre-judged here. Control characters and non-ASCII bytes, which
+ * no transport can encode, are rejected outright. This mirrors the stance taken for values in
  * [requireValidHeaderValues].
  *
  * @return the trimmed, validated name
- * @throws IllegalArgumentException if the trimmed name is blank or contains a control character
+ * @throws IllegalArgumentException if the trimmed name is blank, or contains a control character or
+ *   a non-ASCII byte
  */
 @JvmSynthetic
 internal fun requireValidHeaderName(rawName: String): String {
@@ -48,9 +51,10 @@ internal fun requireValidHeaderName(rawName: String): String {
     require(trimmed.isNotEmpty()) { "Header name must not be blank." }
     trimmed.forEach { ch ->
         require(!isProhibitedInName(ch.code)) {
-            "Header name '${escapeControlCharacters(rawName)}' must not contain control characters " +
-                "(carriage return, line feed, NUL, or other C0/DEL bytes); " +
-                "such characters enable request/header splitting."
+            "Header name '${escapeControlCharacters(rawName)}' must be printable ASCII: it must not " +
+                "contain control characters (carriage return, line feed, NUL, or other C0/DEL bytes) " +
+                "or non-ASCII bytes; control characters enable request/header splitting, and no " +
+                "reference transport can encode a non-ASCII name."
         }
     }
     return trimmed
@@ -70,12 +74,19 @@ internal fun requireValidHeaderName(rawName: String): String {
  * broader control-character set closes the same splitting/injection surface the name check does
  * while staying narrower than the strict field-value grammar.
  *
- * Non-ASCII (for example UTF-8) bytes are NOT rejected — that is the conservative stance shared
- * with the name check: a value some transports accept is not refused at the model layer. [name]
- * only labels the error message; the value itself is never echoed, so a secret or oversized value
- * is not leaked into a log line.
+ * Non-ASCII bytes (code point `>= 0x80`) are rejected as well, matching OkHttp's field-value
+ * grammar (HTAB plus printable ASCII `0x20`–`0x7E`). The shipped transports are not unanimous here
+ * — the JDK `HttpClient` will serialise obs-text (`0x80`+) on the wire — so the model deliberately
+ * takes the *most restrictive* shipped transport's stance: a value it accepts is sendable by every
+ * shipped transport, and a value only some can encode is refused at construction rather than
+ * silently dropped by the stricter one mid-dispatch. This governs **outbound, caller-set** values;
+ * a header parsed from an already-received response takes the lenient
+ * [requireValidInboundHeaderValue] path (via `Headers.Builder.addUnsafeNonAscii`), which preserves
+ * the obs-text a server may legitimately send. [name] only labels the error message; the value
+ * itself is never echoed, so a secret or oversized value is not leaked into a log line.
  *
- * @throws IllegalArgumentException if any value contains a prohibited control character
+ * @throws IllegalArgumentException if any value contains a prohibited control character or a
+ *   non-ASCII byte
  */
 @JvmSynthetic
 internal fun requireValidHeaderValues(
@@ -85,22 +96,61 @@ internal fun requireValidHeaderValues(
     values.forEach { value ->
         value.forEach { ch ->
             require(!isProhibitedInValue(ch.code)) {
-                "Header value for '$name' must not contain control characters (carriage return, " +
-                    "line feed, NUL, or other C0/DEL bytes, except horizontal tab); " +
-                    "such characters enable request/header splitting."
+                "Header value for '$name' must be ASCII: it must not contain control characters " +
+                    "(carriage return, line feed, NUL, or other C0/DEL bytes, except horizontal tab) " +
+                    "or non-ASCII bytes; control characters enable request/header splitting, and " +
+                    "OkHttp's field-value grammar rejects a non-ASCII value."
             }
         }
     }
 }
 
-/** Whether [code] is a control character prohibited in a header name — the full C0 range and DEL. */
-private fun isProhibitedInName(code: Int): Boolean = code <= LAST_C0_CONTROL || code == DEL_CONTROL
+/**
+ * Validates the [value] of an *inbound* (response) header [name] leniently. Like OkHttp's
+ * `addUnsafeNonAscii`/`addLenient` read path, the ASCII restriction is dropped so **non-ASCII bytes
+ * are permitted**; unlike that path — which does no value validation at all — control characters are
+ * still rejected here as defence-in-depth against a misbehaving transport (a stray CR/LF/NUL has no
+ * place in a parsed field-value). RFC 7230 explicitly allows obs-text (`0x80`+) in field-content,
+ * and a server legitimately puts Latin-1 bytes in values such as a `Content-Disposition` filename;
+ * applying the stricter outbound grammar of [requireValidHeaderValues] to a response would silently
+ * strip those headers. [name] only labels the error message; the value itself is never echoed.
+ *
+ * @throws IllegalArgumentException if [value] contains a prohibited control character
+ */
+@JvmSynthetic
+internal fun requireValidInboundHeaderValue(
+    name: String,
+    value: String,
+) {
+    value.forEach { ch ->
+        require(!isProhibitedInInboundValue(ch.code)) {
+            "Inbound header value for '$name' must not contain control characters (carriage " +
+                "return, line feed, NUL, or other C0/DEL bytes, except horizontal tab)."
+        }
+    }
+}
 
 /**
- * Whether [code] is a control character prohibited in a header value — the same set as for a name,
- * minus horizontal tab (`0x09`), which RFC 7230 permits as field-value whitespace.
+ * Whether [code] is prohibited in a header name — the full C0 range, DEL, and every non-ASCII byte
+ * (`>= 0x80`). `code >= DEL_CONTROL` covers DEL (`0x7F`) and all of `0x80`+ in one comparison.
  */
-private fun isProhibitedInValue(code: Int): Boolean =
+private fun isProhibitedInName(code: Int): Boolean = code <= LAST_C0_CONTROL || code >= DEL_CONTROL
+
+/**
+ * Whether [code] is prohibited in an outbound header value — the same set as for a name, minus
+ * horizontal tab (`0x09`), which RFC 7230 permits as field-value whitespace. Shared with [MediaType]
+ * so a media type's rendered form and the header value it becomes agree on the accepted byte set.
+ */
+@JvmSynthetic
+internal fun isProhibitedInValue(code: Int): Boolean =
+    (code <= LAST_C0_CONTROL && code != HORIZONTAL_TAB) || code >= DEL_CONTROL
+
+/**
+ * Whether [code] is prohibited in an *inbound* header value: control characters (`0x00`–`0x1F`
+ * except HTAB, plus DEL `0x7F`) are still rejected, but non-ASCII bytes (`0x80`+) are permitted —
+ * the obs-text a server may legitimately send. See [requireValidInboundHeaderValue].
+ */
+private fun isProhibitedInInboundValue(code: Int): Boolean =
     (code <= LAST_C0_CONTROL && code != HORIZONTAL_TAB) || code == DEL_CONTROL
 
 /**
@@ -126,7 +176,12 @@ private const val HORIZONTAL_TAB: Int = 0x09
 /** Highest code point in the C0 control range (US, `0x1F`); everything at or below is illegal in a name. */
 private const val LAST_C0_CONTROL: Int = 0x1F
 
-/** The DEL control character (`0x7F`), the lone control code above the C0 range. */
+/**
+ * The DEL control character (`0x7F`). Used as the low bound of the rejected high range: for a name
+ * and an outbound value, DEL and every byte above it (the non-ASCII bytes, `0x80`+) are prohibited
+ * (`code >= DEL_CONTROL`); an inbound value rejects only DEL itself (`code == DEL_CONTROL`),
+ * permitting obs-text.
+ */
 private const val DEL_CONTROL: Int = 0x7F
 
 /** Radix for rendering a control character's code point as the hex digits of a `\uXXXX` escape. */
diff --git a/sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/common/Headers.kt b/sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/common/Headers.kt
index ae20c211..800df656 100644
--- a/sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/common/Headers.kt
+++ b/sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/common/Headers.kt
@@ -159,6 +159,36 @@ public data class Headers private constructor(
                 headersMap.computeIfAbsent(canonicalKey(trimmedName)) { mutableListOf() }.addAll(values)
             }
 
+        /**
+         * Adds an inbound (already-received) header, permitting non-ASCII (obs-text) bytes in
+         * [value] that the strict outbound [add] grammar rejects.
+         *
+         * A transport uses this when copying a *response*'s headers into the model. RFC 7230 allows
+         * obs-text (`0x80`+) in a field-value — a server legitimately puts Latin-1 bytes in values
+         * such as a `Content-Disposition` filename — and both reference transports surface such
+         * values on read, so validating them with the outbound grammar would silently drop
+         * legitimate response headers. The name is still validated strictly (control and non-ASCII
+         * bytes rejected); in the value, only the ASCII restriction is relaxed — control characters
+         * are still rejected. Like OkHttp's `addUnsafeNonAscii` this relaxes the value's ASCII
+         * restriction; unlike it (whose `addLenient` validates nothing on the value), the
+         * control-character guard is kept here as defence-in-depth.
+         *
+         * @param name the header name
+         * @param value the header value, which may contain non-ASCII bytes
+         * @return this builder
+         * @throws IllegalArgumentException if [name] is blank or contains a control/non-ASCII byte,
+         *   or [value] contains a prohibited control character
+         */
+        public fun addUnsafeNonAscii(
+            name: String,
+            value: String,
+        ): Builder =
+            apply {
+                val trimmedName = requireValidHeaderName(name)
+                requireValidInboundHeaderValue(trimmedName, value)
+                headersMap.computeIfAbsent(canonicalKey(trimmedName)) { mutableListOf() }.add(value)
+            }
+
         /**
          * Adds a header with the specified typed name and value.
          */
diff --git a/sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/common/HttpHeaderName.kt b/sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/common/HttpHeaderName.kt
index 97dd1c90..f538e3fe 100644
--- a/sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/common/HttpHeaderName.kt
+++ b/sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/common/HttpHeaderName.kt
@@ -24,8 +24,8 @@ import java.util.concurrent.ConcurrentHashMap
  * first caller to intern a given name "wins"; subsequent lookups with different casing
  * yield the same shared instance.
  *
- * Whitespace is trimmed from the input before interning, and the name is validated: a blank name
- * or one carrying an interior control character is rejected (see [fromString]).
+ * Whitespace is trimmed from the input before interning, and the name is validated: a blank name,
+ * or one carrying an interior control character or a non-ASCII byte, is rejected (see [fromString]).
  *
  * Designed for Java 8 bytecode compatibility — no APIs newer than Java 8 are used.
  */
@@ -220,12 +220,14 @@ public class HttpHeaderName private constructor(
          *
          * The name is validated up front by [requireValidHeaderName]: a blank name, or one whose
          * trimmed form contains an interior control character (CR, LF, NUL, or any other C0/DEL
-         * byte), is rejected with an [IllegalArgumentException]. This is the same guard the
-         * String-keyed [Headers.Builder] API applies, so an interned name carried through the typed
-         * header API is guaranteed control-character-free and cannot reach a transport as a
-         * header-splitting vector.
+         * byte) or a non-ASCII byte (code point >= 0x80), is rejected with an
+         * [IllegalArgumentException]. This is the same guard the String-keyed [Headers.Builder] API
+         * applies, so an interned name carried through the typed header API is guaranteed to be
+         * printable ASCII — control-character-free (it cannot reach a transport as a header-splitting
+         * vector) and free of the non-ASCII bytes no reference transport can encode.
          *
-         * @throws IllegalArgumentException if [name] is blank or contains a control character
+         * @throws IllegalArgumentException if [name] is blank, or contains a control character or a
+         *   non-ASCII byte
          */
         @JvmStatic
         public fun fromString(name: String): HttpHeaderName {
diff --git a/sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/common/MediaType.kt b/sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/common/MediaType.kt
index b87cf7a9..145ebc8a 100644
--- a/sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/common/MediaType.kt
+++ b/sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/common/MediaType.kt
@@ -24,10 +24,13 @@ import java.util.Locale
  * (i.e. the bare accept-anything form). Half-wildcard combinations follow normal
  * validation rules.
  *
- * The type, subtype, and parameter keys/values may not contain ASCII control characters — CR, LF,
+ * The type, subtype, and parameter keys/values may not contain an ASCII control character — CR, LF,
  * the rest of the C0 range, and DEL, with HTAB the one permitted exception (RFC 7230 allows it in a
- * `quoted-string`); construction throws [IllegalArgumentException] otherwise. This keeps any of them
- * from injecting an extra header line wherever the media type is rendered into a header.
+ * `quoted-string`) — nor a non-ASCII byte (`0x80`+); construction throws [IllegalArgumentException]
+ * otherwise. Rejecting control characters keeps any of them from injecting an extra header line
+ * wherever the media type is rendered into a header; rejecting non-ASCII keeps the rendered form
+ * within the strict outbound header-value grammar (it shares [isProhibitedInValue] with header-value
+ * validation), so a media type can always be set as a `Content-Type` header without a late rejection.
  *
  * @property type The primary type (e.g., "application", "text").
  * @property subtype The subtype (e.g., "json", "plain").
@@ -42,16 +45,17 @@ public data class MediaType private constructor(
 ) {
     init {
         // A media type is routinely rendered into a header (its own Content-Type, or a multipart
-        // boundary / part Content-Type). Reject control characters in the type, subtype, and every
-        // parameter key/value so none of them can smuggle a CR/LF — and thus an extra header line —
-        // into that output. Validating in the constructor covers every construction path (of, parse,
-        // and copy).
+        // boundary / part Content-Type). Reject control characters and non-ASCII bytes in the type,
+        // subtype, and every parameter key/value so none of them can smuggle a CR/LF — and thus an
+        // extra header line — into that output, and so the rendered form always satisfies the strict
+        // outbound header-value grammar (see isHeaderUnsafe). Validating in the constructor covers
+        // every construction path (of, parse, and copy).
         require(type.none(::isHeaderUnsafe) && subtype.none(::isHeaderUnsafe)) {
-            "Media type must not contain control characters: \"$type/$subtype\""
+            "Media type must not contain control characters or non-ASCII bytes: \"$type/$subtype\""
         }
         parameters.forEach { (key, value) ->
             require(key.none(::isHeaderUnsafe) && value.none(::isHeaderUnsafe)) {
-                "Media type parameter must not contain control characters: \"$key\""
+                "Media type parameter must not contain control characters or non-ASCII bytes: \"$key\""
             }
         }
     }
@@ -144,23 +148,20 @@ public data class MediaType private constructor(
             ch in TOKEN_SPECIALS
 
     /**
-     * True if [ch] is an ASCII control character disallowed inside a header value — any C0 control
-     * (code < 0x20) or DEL (0x7F), **except** HTAB, which RFC 7230 permits in a `quoted-string`.
+     * True if [ch] is disallowed inside a header value, delegating to the shared outbound
+     * header-value predicate ([isProhibitedInValue]) so a media type's rendered form and the
+     * `Content-Type` header value it becomes agree on the accepted byte set: every C0 control and
+     * DEL (except HTAB, which RFC 7230 permits in a `quoted-string`), plus every non-ASCII byte.
      * Rejecting these keeps a parameter from carrying a CR/LF (and thus an extra header line)
-     * wherever this media type is rendered, while still allowing the tabs the wire form supports.
+     * wherever this media type is rendered, and from carrying a byte the strict outbound grammar
+     * would reject once the type is set as a `Content-Type` header.
      */
-    private fun isHeaderUnsafe(ch: Char): Boolean = ch != '\t' && (ch.code < FIRST_PRINTABLE_ASCII || ch.code == DEL)
+    private fun isHeaderUnsafe(ch: Char): Boolean = isProhibitedInValue(ch.code)
 
     public companion object {
         /** RFC 7230 §3.2.6 `tchar` punctuation (the non-alphanumeric members of the token set). */
         private const val TOKEN_SPECIALS: String = "!#$%&'*+-.^_`|~"
 
-        /** Code point of the first printable ASCII character (space); everything below is a C0 control. */
-        private const val FIRST_PRINTABLE_ASCII: Int = 0x20
-
-        /** Code point of the ASCII DEL control character. */
-        private const val DEL: Int = 0x7F
-
         /**
          * Constructs a [MediaType] from explicit parts. All inputs are lower-cased and
          * validated: [type] and [subtype] must be non-blank, and a wildcard [type] is only
diff --git a/sdk-core/src/main/kotlin/org/dexpace/sdk/core/instrumentation/DroppedHeaderLogging.kt b/sdk-core/src/main/kotlin/org/dexpace/sdk/core/instrumentation/DroppedHeaderLogging.kt
new file mode 100644
index 00000000..a61b11a4
--- /dev/null
+++ b/sdk-core/src/main/kotlin/org/dexpace/sdk/core/instrumentation/DroppedHeaderLogging.kt
@@ -0,0 +1,48 @@
+/*
+ * Copyright (c) 2026 dexpace and Omar Aljarrah
+ *
+ * Licensed under the MIT License. See LICENSE in the project root.
+ * SPDX-License-Identifier: MIT
+ */
+
+package org.dexpace.sdk.core.instrumentation
+
+/**
+ * Policy for how a transport logs a request header it drops because it cannot encode it.
+ *
+ * A transport drops a caller-set header when the header is valid at the SDK model layer (printable
+ * ASCII, no control characters) but is rejected by the transport's stricter wire grammar — for
+ * example a name carrying an interior space, which the RFC 7230 `token` rule OkHttp and the JDK
+ * client enforce forbids. The header never reaches the wire, so the drop is worth surfacing; the
+ * question is how loudly, since a hot path re-setting the same non-encodable header would otherwise
+ * emit a log line on every request.
+ *
+ * This policy controls only the drop of a **model-valid but transport-rejected** header. The
+ * pre-drop of a known restricted header the transport recomputes (`Host`, `Content-Length`, …)
+ * always logs at verbose regardless of this setting.
+ *
+ * Note that the log *level* remains the operator's to filter through their SLF4J backend; this
+ * policy chooses the level (and dedup) the transport emits at, which a log level alone cannot
+ * express.
+ */
+public enum class DroppedHeaderLogging {
+    /**
+     * Log every dropped header at `WARN`, on every request. Loudest — a per-request audit trail of
+     * exactly which requests lost which headers, at the cost of high volume on a hot path.
+     */
+    PER_OCCURRENCE,
+
+    /**
+     * Log the **first** drop of a given header name (per transport instance) at `WARN`, and every
+     * subsequent drop of that same name at verbose. The default: a caller learns a header is being
+     * dropped without the log being flooded when the same header is re-set on a hot path. A
+     * different header name still gets its own one-time `WARN`.
+     */
+    ONCE_PER_HEADER,
+
+    /**
+     * Log every dropped header at verbose only; never at `WARN`. Quietest — the loss is off by
+     * default and visible only when the transport's log category is at debug/verbose.
+     */
+    VERBOSE_ONLY,
+}
diff --git a/sdk-core/src/main/kotlin/org/dexpace/sdk/core/pagination/AsyncPaginator.kt b/sdk-core/src/main/kotlin/org/dexpace/sdk/core/pagination/AsyncPaginator.kt
index 6acd1aba..b0be8dec 100644
--- a/sdk-core/src/main/kotlin/org/dexpace/sdk/core/pagination/AsyncPaginator.kt
+++ b/sdk-core/src/main/kotlin/org/dexpace/sdk/core/pagination/AsyncPaginator.kt
@@ -57,8 +57,9 @@ import java.util.function.Consumer
  * A page whose strategy `parse(...)` throws is never built — its response is closed inline on
  * that exceptional path before the result future completes exceptionally. Strategies MUST read
  * everything they need synchronously inside `parse(...)` and MUST NOT retain the response or its
- * body past the call. A page's raw [Response]/body is valid only while that page is being
- * delivered (see [forEachPageAsync]); [Page.items] and the derived metadata outlive close.
+ * body past the call. A page's raw [Response] is open only while that page is being delivered (see
+ * [forEachPageAsync]) — its body is typically already drained by the strategy, so only per-page
+ * metadata is reliably readable there; [Page.items] and the derived metadata outlive close.
  *
  * ## Consumer threading
  *
@@ -118,7 +119,7 @@ import java.util.function.Consumer
  * @property asyncHttpClient Async transport used to execute each page request.
  * @property initialRequest Request used to fetch the first page; also passed to the strategy
  *   as a template for building subsequent page requests.
- * @property strategy Strategy that parses each response into a [Page].
+ * @property strategy Strategy that parses each response into a [PageInfo].
  * @property maxPages Safety cap on the total number of pages (HTTP exchanges) the walk will
  *   fetch. Defaults to `Long.MAX_VALUE` (unbounded). Must be positive.
  */
@@ -131,7 +132,7 @@ public class AsyncPaginator<T>
         private val maxPages: Long = Long.MAX_VALUE,
     ) {
         init {
-            require(maxPages > 0L) { "maxPages must be positive, was $maxPages" }
+            requirePositiveMaxPages(maxPages)
         }
 
         /**
@@ -193,10 +194,13 @@ public class AsyncPaginator<T>
 
         /**
          * Walks every page, invoking [consumer] for each [Page] in server-defined order. The page
-         * passed to [consumer] is **live**: its raw [Page.response]/body is open and readable for
-         * the duration of the callback. The driver closes that response immediately after `accept`
-         * returns, so the consumer MUST NOT retain the page or read its raw body afterwards;
-         * [Page.items] and the derived metadata (status code, headers, request) survive close.
+         * passed to [consumer] is **live**: its [Page.response] is still open for the duration of
+         * the callback, so per-page metadata (status, headers) is readable. The body, however, is
+         * single-use and every shipped strategy drains it inside `parse(...)` to materialize
+         * [Page.items], so it is typically already consumed — do not rely on re-reading it. The
+         * driver closes the response immediately after `accept` returns, so the consumer MUST NOT
+         * retain the page; [Page.items] and the derived metadata (status code, headers, request)
+         * survive close.
          *
          * Cancellation, executor, and ordering semantics are identical to [forEachAsync].
          *
@@ -475,13 +479,7 @@ public class AsyncPaginator<T>
                         // rather than NPE on the parse below.
                         error("AsyncHttpClient.executeAsync completed with a null Response")
                     }
-                    val info =
-                        try {
-                            strategy.parse(response, initialRequest)
-                        } catch (t: Throwable) {
-                            response.close() // parse failed → this page never becomes a Page; release it now
-                            throw t
-                        }
+                    val info = strategy.parseOrClose(response, initialRequest)
                     ParsedPage(Page(response, info.items), info.nextRequest)
                 }
             }
diff --git a/sdk-core/src/main/kotlin/org/dexpace/sdk/core/pagination/CloseablePages.kt b/sdk-core/src/main/kotlin/org/dexpace/sdk/core/pagination/CloseablePages.kt
index d954372b..e34179a1 100644
--- a/sdk-core/src/main/kotlin/org/dexpace/sdk/core/pagination/CloseablePages.kt
+++ b/sdk-core/src/main/kotlin/org/dexpace/sdk/core/pagination/CloseablePages.kt
@@ -8,10 +8,8 @@
 package org.dexpace.sdk.core.pagination
 
 import java.io.IOException
-import java.util.Spliterator
-import java.util.Spliterators
+import java.io.UncheckedIOException
 import java.util.stream.Stream
-import java.util.stream.StreamSupport
 
 /**
  * Auto-closing, single-use page-level view over a lazy pagination walk.
@@ -72,11 +70,36 @@ public class CloseablePages<T>
             }
         }
 
-        /** Sequential, ORDERED, unknown-size stream over the same auto-closing iteration. */
+        /**
+         * Sequential, ORDERED, unknown-size stream over the same auto-closing iteration.
+         *
+         * The stream registers [close] as its close handler, so a short-circuiting terminal
+         * (`findFirst`, `limit`, `anyMatch`) that abandons the stream without exhausting it still
+         * releases the page left held open — **provided the stream is closed**. Wrap it in
+         * try-with-resources (Java) / `use { }` (Kotlin):
+         *
+         * ```java
+         * try (Stream<Page<X>> s = pages.stream()) { s.findFirst(); } // closes the held page
+         * ```
+         *
+         * Equivalently, this [CloseablePages] is itself the close handle — closing it also releases
+         * the held page: `byPage().use { it.stream().findFirst() }`.
+         *
+         * [Stream.close] declares no checked exception, so the `IOException` from [close] is
+         * rethrown wrapped in an [UncheckedIOException] (as the JDK's own `Files.lines` does) rather
+         * than sneaky-thrown undeclared — a Java caller can then `catch (UncheckedIOException)` at
+         * the try-with-resources site instead of an uncatchable-by-type checked exception.
+         */
         public fun stream(): Stream<Page<T>> =
-            StreamSupport.stream(
-                Spliterators.spliteratorUnknownSize(iterator(), Spliterator.ORDERED),
-                false,
+            orderedStream(
+                iterator(),
+                Runnable {
+                    try {
+                        close()
+                    } catch (e: IOException) {
+                        throw UncheckedIOException(e)
+                    }
+                },
             )
 
         /**
diff --git a/sdk-core/src/main/kotlin/org/dexpace/sdk/core/pagination/PageWalker.kt b/sdk-core/src/main/kotlin/org/dexpace/sdk/core/pagination/PageWalker.kt
index 04636ed0..254612c5 100644
--- a/sdk-core/src/main/kotlin/org/dexpace/sdk/core/pagination/PageWalker.kt
+++ b/sdk-core/src/main/kotlin/org/dexpace/sdk/core/pagination/PageWalker.kt
@@ -7,17 +7,15 @@
 
 package org.dexpace.sdk.core.pagination
 
-import java.util.Spliterator
-import java.util.Spliterators
 import java.util.stream.Stream
-import java.util.stream.StreamSupport
 
 /**
  * Internal lazy driver shared by [Paginator] and [PagedIterable]. Pulls pages from [source]
  * — which fully encapsulates its own paging state (cursor, next request, link) and returns
  * `null` at end of stream — and stops after [maxPages] pages. Page-lazy: [source] is invoked
- * only when the consumer pulls the next page. Also the single home for the iterator→`Stream`
- * adaptation, so the public wrappers do not each re-implement it.
+ * only when the consumer pulls the next page. Its [itemStream] goes through the shared
+ * [orderedStream] adapter (which the page-level [CloseablePages] view also uses), so the
+ * iterator→`Stream` boilerplate lives in one place rather than in each public wrapper.
  *
  * Page closing is split by view: [items] eager-closes each page before yielding any of its
  * items (so a partial item-consume never strands a connection), while [pages] yields **live**
@@ -32,7 +30,7 @@ internal class PageWalker<T>(
     private val maxPages: Long,
 ) {
     init {
-        require(maxPages > 0L) { "maxPages must be positive, was $maxPages" }
+        requirePositiveMaxPages(maxPages)
     }
 
     /**
@@ -66,11 +64,5 @@ internal class PageWalker<T>(
         }
 
     /** Sequential, ORDERED, unknown-size [Stream] over [items]. */
-    fun itemStream(): Stream<T> = streamOf(items())
-
-    private fun <E> streamOf(iterator: Iterator<E>): Stream<E> =
-        StreamSupport.stream(
-            Spliterators.spliteratorUnknownSize(iterator, Spliterator.ORDERED),
-            false,
-        )
+    fun itemStream(): Stream<T> = orderedStream(items())
 }
diff --git a/sdk-core/src/main/kotlin/org/dexpace/sdk/core/pagination/PagedIterable.kt b/sdk-core/src/main/kotlin/org/dexpace/sdk/core/pagination/PagedIterable.kt
index 74553234..5fabf09a 100644
--- a/sdk-core/src/main/kotlin/org/dexpace/sdk/core/pagination/PagedIterable.kt
+++ b/sdk-core/src/main/kotlin/org/dexpace/sdk/core/pagination/PagedIterable.kt
@@ -47,7 +47,7 @@ public class PagedIterable<T>
         private val maxPages: Long = Long.MAX_VALUE,
     ) : Iterable<T> {
         init {
-            require(maxPages > 0L) { "maxPages must be positive, was $maxPages" }
+            requirePositiveMaxPages(maxPages)
         }
 
         private fun walker(options: PagingOptions): PageWalker<T> {
diff --git a/sdk-core/src/main/kotlin/org/dexpace/sdk/core/pagination/PaginationInternal.kt b/sdk-core/src/main/kotlin/org/dexpace/sdk/core/pagination/PaginationInternal.kt
new file mode 100644
index 00000000..a4cef58a
--- /dev/null
+++ b/sdk-core/src/main/kotlin/org/dexpace/sdk/core/pagination/PaginationInternal.kt
@@ -0,0 +1,75 @@
+/*
+ * Copyright (c) 2026 dexpace and Omar Aljarrah
+ *
+ * Licensed under the MIT License. See LICENSE in the project root.
+ * SPDX-License-Identifier: MIT
+ */
+
+package org.dexpace.sdk.core.pagination
+
+import org.dexpace.sdk.core.http.request.Request
+import org.dexpace.sdk.core.http.response.Response
+import java.util.Spliterator
+import java.util.Spliterators
+import java.util.stream.Stream
+import java.util.stream.StreamSupport
+
+/*
+ * Shared internal helpers for the `pagination` package, factored out so the sync and async
+ * paginators, the walker, and the page-level view do not each re-implement the same one-liners
+ * (and cannot silently diverge on the invariants those one-liners encode).
+ */
+
+/**
+ * Rejects a non-positive [maxPages] with the message the public paginator wrappers share. Each
+ * wrapper checks eagerly in its own `init` so the failure surfaces when the object is built, not
+ * lazily on first walk; centralising the message keeps those checks identical.
+ */
+@JvmSynthetic
+internal fun requirePositiveMaxPages(maxPages: Long) {
+    require(maxPages > 0L) { "maxPages must be positive, was $maxPages" }
+}
+
+/**
+ * Adapts [iterator] into a sequential, ORDERED, unknown-size [Stream] — the single home for the
+ * iterator→`Stream` boilerplate the pagination views share ([PageWalker.itemStream] and
+ * [CloseablePages.stream]). When [onClose] is supplied it is registered as the stream's close
+ * handler, so closing the returned `Stream` (try-with-resources / `use { }`) runs it — the seam
+ * a page-level view uses to release a page a short-circuiting terminal left held open.
+ */
+@JvmSynthetic
+internal fun <E> orderedStream(
+    iterator: Iterator<E>,
+    onClose: Runnable? = null,
+): Stream<E> {
+    val stream: Stream<E> =
+        StreamSupport.stream(
+            Spliterators.spliteratorUnknownSize(iterator, Spliterator.ORDERED),
+            false,
+        )
+    return if (onClose == null) stream else stream.onClose(onClose)
+}
+
+/**
+ * Parses [response] with this strategy, closing [response] and rethrowing if `parse` fails. A page
+ * whose parse throws never becomes a [Page], so nothing else will ever close its live response —
+ * releasing it here is the close-on-parse-failure invariant both paginators must keep identical.
+ */
+@JvmSynthetic
+internal fun <T> PaginationStrategy<T>.parseOrClose(
+    response: Response,
+    initialRequest: Request,
+): PageInfo<T> =
+    try {
+        parse(response, initialRequest)
+    } catch (t: Throwable) {
+        // parse failed → this page never becomes a Page; release it now. A failure closing the
+        // response must not mask the parse failure that is actually propagating, so attach it as
+        // suppressed rather than letting it replace `t`.
+        try {
+            response.close()
+        } catch (closeFailure: Throwable) {
+            t.addSuppressed(closeFailure)
+        }
+        throw t
+    }
diff --git a/sdk-core/src/main/kotlin/org/dexpace/sdk/core/pagination/PaginationStrategy.kt b/sdk-core/src/main/kotlin/org/dexpace/sdk/core/pagination/PaginationStrategy.kt
index 6a400fb9..6a3b93c8 100644
--- a/sdk-core/src/main/kotlin/org/dexpace/sdk/core/pagination/PaginationStrategy.kt
+++ b/sdk-core/src/main/kotlin/org/dexpace/sdk/core/pagination/PaginationStrategy.kt
@@ -11,7 +11,7 @@ import org.dexpace.sdk.core.http.request.Request
 import org.dexpace.sdk.core.http.response.Response
 
 /**
- * Parses a [Response] into a [Page] and uses the [initialRequest] as a template for
+ * Parses a [Response] into a [PageInfo] and uses the [initialRequest] as a template for
  * building the next page's request.
  *
  * `PaginationStrategy` is the seam that lets the [Paginator] support any wire convention —
@@ -34,7 +34,7 @@ import org.dexpace.sdk.core.http.response.Response
  */
 public fun interface PaginationStrategy<T> {
     /**
-     * Parses [response] into a [Page] of items. Uses [initialRequest] as a template for
+     * Parses [response] into a [PageInfo] of items. Uses [initialRequest] as a template for
      * the next page's URL/headers/method/body, mutating only what the wire convention
      * demands (e.g. swapping the cursor query param, replacing the URL with a
      * `Link: rel="next"` value).
diff --git a/sdk-core/src/main/kotlin/org/dexpace/sdk/core/pagination/Paginator.kt b/sdk-core/src/main/kotlin/org/dexpace/sdk/core/pagination/Paginator.kt
index 63d80a38..a0772ff5 100644
--- a/sdk-core/src/main/kotlin/org/dexpace/sdk/core/pagination/Paginator.kt
+++ b/sdk-core/src/main/kotlin/org/dexpace/sdk/core/pagination/Paginator.kt
@@ -78,7 +78,7 @@ import java.util.stream.Stream
  * @property httpClient Transport used to execute each page request.
  * @property initialRequest Request used to fetch the first page; also passed to the
  *   strategy as a template for building subsequent page requests.
- * @property strategy Strategy that parses each response into a [Page].
+ * @property strategy Strategy that parses each response into a [PageInfo].
  * @property maxPages Safety cap on the total number of pages (HTTP exchanges) the iterator
  *   will fetch. Defaults to `Long.MAX_VALUE` (unbounded). Must be positive.
  */
@@ -91,7 +91,7 @@ public class Paginator<T>
         private val maxPages: Long = Long.MAX_VALUE,
     ) {
         init {
-            require(maxPages > 0L) { "maxPages must be positive, was $maxPages" }
+            requirePositiveMaxPages(maxPages)
         }
 
         /** Lazy iterable over all items across all pages. Each call restarts pagination. */
@@ -114,13 +114,7 @@ public class Paginator<T>
                     val request = nextRequest ?: return@PageWalker null
                     nextRequest = null
                     val response: Response = httpClient.execute(request)
-                    val info: PageInfo<T> =
-                        try {
-                            strategy.parse(response, initialRequest)
-                        } catch (t: Throwable) {
-                            response.close() // parse failed → this page never becomes a Page; release it now
-                            throw t
-                        }
+                    val info: PageInfo<T> = strategy.parseOrClose(response, initialRequest)
                     nextRequest = info.nextRequest
                     Page(response, info.items)
                 },
diff --git a/sdk-core/src/test/kotlin/org/dexpace/sdk/core/http/common/HeadersTest.kt b/sdk-core/src/test/kotlin/org/dexpace/sdk/core/http/common/HeadersTest.kt
index 29f989c5..886f6131 100644
--- a/sdk-core/src/test/kotlin/org/dexpace/sdk/core/http/common/HeadersTest.kt
+++ b/sdk-core/src/test/kotlin/org/dexpace/sdk/core/http/common/HeadersTest.kt
@@ -281,15 +281,57 @@ class HeadersTest {
                 .add("X-Plain", "hello world")
                 .set("Authorization", "Bearer abc.def-ghi")
                 .add(HttpHeaderName.SET_COOKIE, "id=42; Path=/")
-                // Horizontal tab is the one permitted control character, and UTF-8 is not a control
-                // character at all, so both pass the conservative check.
-                .set("X-Unicode", "café\tvalue")
+                // Horizontal tab is the one permitted control character; printable ASCII plus tab
+                // passes the check.
+                .set("X-Tabbed", "left\tright")
                 .build()
 
         assertEquals("hello world", headers.get("X-Plain"))
         assertEquals("Bearer abc.def-ghi", headers.get("Authorization"))
         assertEquals("id=42; Path=/", headers.get(HttpHeaderName.SET_COOKIE))
-        assertEquals("café\tvalue", headers.get("X-Unicode"))
+        assertEquals("left\tright", headers.get("X-Tabbed"))
+    }
+
+    @Test
+    fun `value validation rejects a non-ASCII byte`() {
+        // RFC 7230 field-values are ASCII, and both reference transports reject a non-ASCII value,
+        // so the model refuses it at construction rather than accepting a value no transport can put
+        // on the wire (previously it was permitted at the model layer and dropped mid-dispatch).
+        val eAcute = 233.toChar() // 'é' (U+00E9): valid UTF-8, non-ASCII
+        assertFailsWith<IllegalArgumentException> { Headers.builder().add("X-Foo", "caf" + eAcute) }
+        assertFailsWith<IllegalArgumentException> { Headers.builder().set("X-Foo", "caf" + eAcute) }
+    }
+
+    @Test
+    fun `addUnsafeNonAscii preserves a non-ASCII value the outbound grammar would reject`() {
+        // The inbound path exists so a response header carrying legal obs-text (e.g. a Latin-1
+        // Content-Disposition filename) survives, where the strict outbound add/set would reject it.
+        val eAcute = 233.toChar() // 'é' (U+00E9)
+        val value = "attachment; filename=\"caf" + eAcute + ".pdf\""
+        // Guard: the outbound grammar still rejects the same value, so this is a genuine relaxation.
+        assertFailsWith<IllegalArgumentException> { Headers.builder().add("Content-Disposition", value) }
+
+        val headers = Headers.builder().addUnsafeNonAscii("Content-Disposition", value).build()
+        assertEquals(value, headers.get("Content-Disposition"))
+    }
+
+    @Test
+    fun `addUnsafeNonAscii permits horizontal tab but rejects control characters in the value`() {
+        val tabbed = Headers.builder().addUnsafeNonAscii("X-Tabbed", "left\tright").build()
+        assertEquals("left\tright", tabbed.get("X-Tabbed"))
+
+        // A CR/LF/NUL/DEL in a parsed value is still refused — defence-in-depth against a
+        // misbehaving transport, matching the pre-existing outbound guard.
+        assertFailsWith<IllegalArgumentException> { Headers.builder().addUnsafeNonAscii("X-Bad", "a\rb") }
+        assertFailsWith<IllegalArgumentException> { Headers.builder().addUnsafeNonAscii("X-Bad", "a\u0000b") }
+        assertFailsWith<IllegalArgumentException> { Headers.builder().addUnsafeNonAscii("X-Bad", "a\u007Fb") }
+    }
+
+    @Test
+    fun `addUnsafeNonAscii still validates the name strictly`() {
+        val eAcute = 233.toChar() // 'é' (U+00E9)
+        assertFailsWith<IllegalArgumentException> { Headers.builder().addUnsafeNonAscii("X-Caf" + eAcute, "v") }
+        assertFailsWith<IllegalArgumentException> { Headers.builder().addUnsafeNonAscii("X-Evil\nInjected", "v") }
     }
 
     @Test
@@ -370,6 +412,15 @@ class HeadersTest {
         }
     }
 
+    @Test
+    fun `add and set reject a name containing a non-ASCII byte`() {
+        // RFC 7230 field-names are ASCII, and no reference transport can encode a non-ASCII name, so
+        // it is refused at construction rather than accepted and dropped mid-dispatch.
+        val eAcute = 233.toChar() // 'é' (U+00E9): non-ASCII, not an RFC 7230 token char
+        assertFailsWith<IllegalArgumentException> { Headers.builder().add("X-Caf" + eAcute, "v") }
+        assertFailsWith<IllegalArgumentException> { Headers.builder().set("X-Caf" + eAcute, listOf("v")) }
+    }
+
     @Test
     fun `add rejects a blank name`() {
         assertFailsWith<IllegalArgumentException> {
diff --git a/sdk-core/src/test/kotlin/org/dexpace/sdk/core/http/common/HttpHeaderNameTest.kt b/sdk-core/src/test/kotlin/org/dexpace/sdk/core/http/common/HttpHeaderNameTest.kt
index df0979f7..a4353d29 100644
--- a/sdk-core/src/test/kotlin/org/dexpace/sdk/core/http/common/HttpHeaderNameTest.kt
+++ b/sdk-core/src/test/kotlin/org/dexpace/sdk/core/http/common/HttpHeaderNameTest.kt
@@ -104,6 +104,15 @@ class HttpHeaderNameTest {
         assertFailsWith<IllegalArgumentException> { HttpHeaderName.fromString("X-Evil" + del + "Injected") }
     }
 
+    @Test
+    fun `fromString rejects a name with a non-ASCII byte`() {
+        // The typed API shares Headers.Builder's tightened validation: RFC 7230 field-names are
+        // ASCII and no reference transport can encode a non-ASCII name, so it is refused at
+        // construction rather than accepted at the model layer and dropped mid-dispatch.
+        val eAcute = 233.toChar() // 'é' (U+00E9): non-ASCII
+        assertFailsWith<IllegalArgumentException> { HttpHeaderName.fromString("X-Caf" + eAcute) }
+    }
+
     @Test
     fun `concurrent fromString calls converge on a single interned instance`() {
         // Pick a name not used by any well-known constant or other test so this race
diff --git a/sdk-core/src/test/kotlin/org/dexpace/sdk/core/http/common/MediaTypeTest.kt b/sdk-core/src/test/kotlin/org/dexpace/sdk/core/http/common/MediaTypeTest.kt
index e22d8314..dc52befe 100644
--- a/sdk-core/src/test/kotlin/org/dexpace/sdk/core/http/common/MediaTypeTest.kt
+++ b/sdk-core/src/test/kotlin/org/dexpace/sdk/core/http/common/MediaTypeTest.kt
@@ -223,6 +223,23 @@ class MediaTypeTest {
         }
     }
 
+    @Test
+    fun `of rejects a non-ASCII byte in a parameter value`() {
+        // The rendered media type becomes a Content-Type header value, which the strict outbound
+        // grammar restricts to ASCII; a non-ASCII parameter must be refused at construction rather
+        // than accepted and then rejected when set as a header. The message names non-ASCII too, not
+        // just control characters.
+        val eAcute = 233.toChar() // 'é' (U+00E9): non-ASCII, not a control character
+        val ex =
+            assertFailsWith<IllegalArgumentException> {
+                MediaType.of("text", "plain", mapOf("title" to "caf" + eAcute))
+            }
+        assertTrue(
+            ex.message?.contains("non-ASCII") == true,
+            "message must mention non-ASCII, not only control characters: ${ex.message}",
+        )
+    }
+
     // ---- charset accessor -------------------------------------------------------
 
     @Test
diff --git a/sdk-core/src/test/kotlin/org/dexpace/sdk/core/pagination/AsyncPaginatorTest.kt b/sdk-core/src/test/kotlin/org/dexpace/sdk/core/pagination/AsyncPaginatorTest.kt
index e261570b..4fffa22f 100644
--- a/sdk-core/src/test/kotlin/org/dexpace/sdk/core/pagination/AsyncPaginatorTest.kt
+++ b/sdk-core/src/test/kotlin/org/dexpace/sdk/core/pagination/AsyncPaginatorTest.kt
@@ -31,6 +31,7 @@ import kotlin.test.Test
 import kotlin.test.assertEquals
 import kotlin.test.assertFailsWith
 import kotlin.test.assertFalse
+import kotlin.test.assertSame
 import kotlin.test.assertTrue
 
 class AsyncPaginatorTest {
@@ -828,6 +829,96 @@ class AsyncPaginatorTest {
         assertTrue(ex.cause is IOException, "cause was ${ex.cause}")
         assertEquals(listOf("a", "b"), seen, "items are delivered before the close failure surfaces")
     }
+
+    @Test
+    fun `a throwing sink over a throwing close surfaces the sink failure, not the close error`() {
+        // drainPage intersection: the sink throws AND the page's close() throws. The sink cause is
+        // recorded first and settles the result; the subsequent close IOException must be swallowed
+        // (result already done), so the caller sees the sink failure — not the close error. The two
+        // inputs were only ever tested in isolation before.
+        val client = StubAsyncHttpClient()
+        client.on("https://api.example.com/items") { req -> closeThrowingResponse(req, "a,b") }
+        val paginator = AsyncPaginator(client, initialRequest(), strategy())
+
+        val sinkBoom = RuntimeException("sink exploded")
+        val ex =
+            assertFailsWith<ExecutionException> {
+                paginator.forEachAsync { throw sinkBoom }.get(5, TimeUnit.SECONDS)
+            }
+        assertSame(sinkBoom, ex.cause, "the sink failure must be the surfaced cause, not the close IOException")
+    }
+
+    @Test
+    fun `a throwing close on a staged page dropped by a rejected re-dispatch is swallowed`() {
+        // closeStagedPageQuietly on the rejected-re-dispatch path: mirrors the rejected-re-dispatch
+        // leak test, but the staged page's close() throws. resume()'s catch calls
+        // closeStagedPageQuietly(), which must swallow the close IOException so the
+        // RejectedExecutionException — the real failure — is what the result carries.
+        val page1Transport = CompletableFuture<Response>()
+        val client =
+            AsyncHttpClient { req ->
+                if (req.url.toString() == "https://api.example.com/items") {
+                    page1Transport
+                } else {
+                    CompletableFuture.completedFuture(textResponse(req, "c,d"))
+                }
+            }
+        val paginator = AsyncPaginator(client, initialRequest(), strategy())
+
+        // allowed = 1: the initial dispatch runs inline; the re-entry dispatch is rejected.
+        val executor = RejectAfterNExecutor(allowed = 1)
+        val result = paginator.forEachAsync({ }, executor)
+        assertFalse(result.isDone, "walk must be suspended on the in-flight page")
+
+        // Staging page 1 triggers the re-entry dispatch, which the executor rejects; the staged
+        // page's throwing close() must not mask the rejection.
+        page1Transport.complete(closeThrowingResponse(initialRequest(), "a,b"))
+
+        val ex =
+            assertFailsWith<ExecutionException> {
+                result.get(5, TimeUnit.SECONDS)
+            }
+        assertTrue(
+            ex.cause is RejectedExecutionException,
+            "the rejection must survive the swallowed close error, cause was ${ex.cause}",
+        )
+    }
+
+    @Test
+    fun `a throwing close on a page dropped by external settle is swallowed, leaving a clean cancellation`() {
+        // closeStagedPageQuietly on the external-settle drop path: mirrors the external-settle
+        // drop-leak test, but the dropped page's close() throws. drainStaged() drops the page via
+        // closeStagedPageQuietly() after the result was cancelled; the close IOException must be
+        // swallowed so the result stays a clean cancellation.
+        val page1Transport = CompletableFuture<Response>()
+        val client =
+            AsyncHttpClient { req ->
+                if (req.url.toString() == "https://api.example.com/items") {
+                    page1Transport
+                } else {
+                    CompletableFuture.completedFuture(textResponse(req, "c,d"))
+                }
+            }
+        val paginator = AsyncPaginator(client, initialRequest(), strategy())
+
+        val executor = ManualExecutor()
+        val pagesDelivered = AtomicInteger(0)
+        val result = paginator.forEachPageAsync({ pagesDelivered.incrementAndGet() }, executor)
+
+        assertTrue(executor.runNext(), "the initial drive task must be queued")
+        assertFalse(result.isDone, "walk must be suspended on the in-flight page")
+
+        // Stage page 1 (retaining its throwing-close response), then settle from outside before it
+        // is drained so drainStaged() drops-and-closes it.
+        page1Transport.complete(closeThrowingResponse(initialRequest(), "a,b"))
+        result.cancel(true)
+
+        assertTrue(executor.runNext(), "the drop task must be queued")
+        executor.runAll()
+
+        assertTrue(result.isCancelled, "external cancellation stands; the swallowed close error must not override it")
+        assertEquals(0, pagesDelivered.get(), "a dropped page must not reach the consumer")
+    }
 }
 
 /**
diff --git a/sdk-core/src/test/kotlin/org/dexpace/sdk/core/pagination/PaginationTestSupport.kt b/sdk-core/src/test/kotlin/org/dexpace/sdk/core/pagination/PaginationTestSupport.kt
index 4aac3830..0eff452f 100644
--- a/sdk-core/src/test/kotlin/org/dexpace/sdk/core/pagination/PaginationTestSupport.kt
+++ b/sdk-core/src/test/kotlin/org/dexpace/sdk/core/pagination/PaginationTestSupport.kt
@@ -135,6 +135,46 @@ private fun closeRecordingBody(
     }
 }
 
+/**
+ * Convenience: a 200 OK response whose body reads normally but whose `close()` always throws an
+ * [java.io.IOException] with [closeMessage]. Lets tests exercise the close-failure paths — the
+ * parse-failure suppression in [parseOrClose] and the stream close handler's `UncheckedIOException`
+ * wrapping in [CloseablePages.stream].
+ */
+internal fun closeFailingResponse(
+    request: Request,
+    body: String = "",
+    closeMessage: String = "close exploded",
+): Response =
+    Response.builder()
+        .request(request)
+        .protocol(Protocol.HTTP_1_1)
+        .status(Status.OK)
+        .headers(Headers.Builder().build())
+        .body(closeFailingBody(body, closeMessage))
+        .build()
+
+private fun closeFailingBody(
+    content: String,
+    closeMessage: String,
+): ResponseBody {
+    val bytes = content.toByteArray(Charsets.UTF_8)
+    return object : ResponseBody() {
+        private var cachedSource: org.dexpace.sdk.core.io.BufferedSource? = null
+
+        override fun mediaType(): org.dexpace.sdk.core.http.common.MediaType? = null
+
+        override fun contentLength(): Long = bytes.size.toLong()
+
+        override fun source(): org.dexpace.sdk.core.io.BufferedSource =
+            cachedSource ?: Io.provider.source(bytes).also { cachedSource = it }
+
+        override fun close() {
+            throw java.io.IOException(closeMessage)
+        }
+    }
+}
+
 /** Install the Okio IoProvider so test response bodies are readable. */
 internal fun installIoProvider() {
     Io.installProvider(OkioIoProvider)
diff --git a/sdk-core/src/test/kotlin/org/dexpace/sdk/core/pagination/PaginatorCloseTest.kt b/sdk-core/src/test/kotlin/org/dexpace/sdk/core/pagination/PaginatorCloseTest.kt
new file mode 100644
index 00000000..e45f1320
--- /dev/null
+++ b/sdk-core/src/test/kotlin/org/dexpace/sdk/core/pagination/PaginatorCloseTest.kt
@@ -0,0 +1,151 @@
+/*
+ * Copyright (c) 2026 dexpace and Omar Aljarrah
+ *
+ * Licensed under the MIT License. See LICENSE in the project root.
+ * SPDX-License-Identifier: MIT
+ */
+
+package org.dexpace.sdk.core.pagination
+
+import org.dexpace.sdk.core.http.request.Method
+import org.dexpace.sdk.core.http.request.Request
+import org.dexpace.sdk.core.http.response.Response
+import java.io.IOException
+import java.io.UncheckedIOException
+import java.util.concurrent.atomic.AtomicInteger
+import kotlin.test.BeforeTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFailsWith
+
+/**
+ * Response-close coverage for the sync [Paginator] that the async mirror already has: the
+ * parse-failure release branch, the page-level early-break release, and the stream short-circuit
+ * regression (the held page is released when a short-circuited page stream is closed).
+ */
+class PaginatorCloseTest {
+    @BeforeTest fun setup() = installIoProvider()
+
+    private fun initialRequest(): Request =
+        Request.builder()
+            .url("https://api.example.com/items")
+            .method(Method.GET)
+            .build()
+
+    private val extractor: (Response) -> CursorResult<String> = { resp ->
+        val body = resp.body!!.source().use { it.readUtf8() }
+        val itemsLine = body.lineSequence().firstOrNull { it.startsWith("items=") } ?: "items="
+        val cursorLine = body.lineSequence().firstOrNull { it.startsWith("cursor=") } ?: "cursor="
+        val itemsRaw = itemsLine.removePrefix("items=")
+        val cursorRaw = cursorLine.removePrefix("cursor=")
+        val items = if (itemsRaw.isEmpty()) emptyList() else itemsRaw.split(",")
+        CursorResult(items, cursorRaw.ifEmpty { null })
+    }
+
+    /** Two-page cursor stub whose page responses record their close count into [closes]. */
+    private fun twoPageClient(closes: AtomicInteger): StubHttpClient =
+        StubHttpClient()
+            .on("https://api.example.com/items") { req ->
+                closeRecordingResponse(req, closes, "items=a,b\ncursor=abc")
+            }
+            .on("https://api.example.com/items?cursor=abc") { req ->
+                closeRecordingResponse(req, closes, "items=c\ncursor=")
+            }
+
+    @Test
+    fun `sync strategy parse failure surfaces and still closes the response`() {
+        // The async mirror is tested twice, but no sync test drove a throwing strategy — deleting
+        // the sync walker() response.close() would strand a connection on every sync parse failure
+        // and leave the build green. A throwing strategy must propagate AND release the response.
+        val closes = AtomicInteger(0)
+        val client = StubHttpClient()
+        client.on("https://api.example.com/items") { req -> closeRecordingResponse(req, closes, "items=a,b") }
+        val failing = PaginationStrategy<String> { _, _ -> throw IOException("parse exploded") }
+        val paginator = Paginator(client, initialRequest(), failing)
+
+        val ex =
+            assertFailsWith<IOException> {
+                paginator.iterateAll().iterator().hasNext()
+            }
+        assertEquals("parse exploded", ex.message)
+        assertEquals(1, closes.get(), "response must be closed even when parse throws")
+    }
+
+    @Test
+    fun `sync parse failure attaches a throwing close as suppressed and does not mask the parse error`() {
+        // When releasing the page whose parse failed, a close() that itself throws must not replace
+        // the parse failure the caller needs to see — it is attached as suppressed instead.
+        val client = StubHttpClient()
+        client.on("https://api.example.com/items") { req -> closeFailingResponse(req, "items=a,b", "close exploded") }
+        val failing = PaginationStrategy<String> { _, _ -> throw IOException("parse exploded") }
+        val paginator = Paginator(client, initialRequest(), failing)
+
+        val ex =
+            assertFailsWith<IOException> {
+                paginator.iterateAll().iterator().hasNext()
+            }
+        assertEquals("parse exploded", ex.message, "the parse failure must propagate, not the close failure")
+        val suppressed = ex.suppressed.toList()
+        assertEquals(1, suppressed.size, "the close failure must be attached as suppressed, not swallowed or masking")
+        assertEquals("close exploded", suppressed.first().message)
+    }
+
+    @Test
+    fun `byPage stream rethrows a failing page close as UncheckedIOException on stream close`() {
+        // Stream.close() declares no checked exception, so a held page whose close() throws IOException
+        // must surface as an UncheckedIOException (not a sneaky-thrown, uncatchable-by-type checked one).
+        val client =
+            StubHttpClient().on("https://api.example.com/items") { req ->
+                closeFailingResponse(req, "items=a,b\ncursor=abc", "held page close exploded")
+            }
+        val paginator = Paginator(client, initialRequest(), CursorPaginationStrategy(extractor, "cursor"))
+
+        val stream = paginator.byPage().stream()
+        stream.findFirst() // pulls page 1, leaves it held open
+        val ex = assertFailsWith<UncheckedIOException> { stream.close() }
+        assertEquals("held page close exploded", ex.cause?.message, "the original IOException must be the cause")
+    }
+
+    @Test
+    fun `byPage early break without use leaves one page open and close releases it`() {
+        // Sync mirror of PagedIterableTest's early-break case, but over a live-Response client:
+        // breaking while page 2 is current leaves it open; view.close() releases it and is idempotent.
+        val closes = AtomicInteger(0)
+        val paginator =
+            Paginator(twoPageClient(closes), initialRequest(), CursorPaginationStrategy(extractor, "cursor"))
+
+        val view = paginator.byPage()
+        val pages = view.iterator()
+        var advanced = 0
+        while (pages.hasNext()) {
+            pages.next()
+            advanced++
+            if (advanced == 2) break // break while page 2 is still current/open
+        }
+        assertEquals(1, closes.get(), "after breaking on page 2 only page 1 (advanced past) is closed")
+
+        view.close()
+        assertEquals(2, closes.get(), "close() must release the still-held page 2")
+
+        view.close()
+        assertEquals(2, closes.get(), "close() is idempotent — no double-close")
+    }
+
+    @Test
+    fun `byPage stream short-circuit releases the held page when the stream is closed`() {
+        // Regression: a short-circuiting terminal (findFirst) pulls one page and abandons the stream
+        // without exhausting it. With the stream's onClose wired to the view, closing the stream
+        // (try-with-resources / use) releases the page left held — no stranded connection.
+        val closes = AtomicInteger(0)
+        val paginator =
+            Paginator(twoPageClient(closes), initialRequest(), CursorPaginationStrategy(extractor, "cursor"))
+
+        val first: Page<String> =
+            paginator.byPage().stream().use { pages ->
+                pages.findFirst().orElseThrow { AssertionError("expected at least one page") }
+            }
+
+        assertEquals(listOf("a", "b"), first.items, "materialized items survive the page close")
+        assertEquals(1, closes.get(), "closing the short-circuited stream must release the held page")
+    }
+}
diff --git a/sdk-transport-jdkhttp/api/sdk-transport-jdkhttp.api b/sdk-transport-jdkhttp/api/sdk-transport-jdkhttp.api
index 5f38060c..661e8c57 100644
--- a/sdk-transport-jdkhttp/api/sdk-transport-jdkhttp.api
+++ b/sdk-transport-jdkhttp/api/sdk-transport-jdkhttp.api
@@ -1,10 +1,11 @@
 public final class org/dexpace/sdk/transport/jdkhttp/JdkHttpTransport : org/dexpace/sdk/core/client/AsyncHttpClient, org/dexpace/sdk/core/client/HttpClient {
 	public static final field Companion Lorg/dexpace/sdk/transport/jdkhttp/JdkHttpTransport$Companion;
-	public synthetic fun <init> (Ljava/net/http/HttpClient;Ljava/time/Duration;ZLjava/util/concurrent/ExecutorService;Lkotlin/jvm/internal/DefaultConstructorMarker;)V
+	public synthetic fun <init> (Ljava/net/http/HttpClient;Ljava/time/Duration;ZLjava/util/concurrent/ExecutorService;Lorg/dexpace/sdk/core/instrumentation/DroppedHeaderLogging;Lkotlin/jvm/internal/DefaultConstructorMarker;)V
 	public static final fun builder ()Lorg/dexpace/sdk/transport/jdkhttp/JdkHttpTransport$Builder;
 	public fun close ()V
 	public static final fun create (Ljava/net/http/HttpClient;)Lorg/dexpace/sdk/transport/jdkhttp/JdkHttpTransport;
 	public static final fun create (Ljava/net/http/HttpClient;Ljava/time/Duration;)Lorg/dexpace/sdk/transport/jdkhttp/JdkHttpTransport;
+	public static final fun create (Ljava/net/http/HttpClient;Ljava/time/Duration;Lorg/dexpace/sdk/core/instrumentation/DroppedHeaderLogging;)Lorg/dexpace/sdk/transport/jdkhttp/JdkHttpTransport;
 	public fun execute (Lorg/dexpace/sdk/core/http/request/Request;)Lorg/dexpace/sdk/core/http/response/Response;
 	public fun executeAsync (Lorg/dexpace/sdk/core/http/request/Request;)Ljava/util/concurrent/CompletableFuture;
 }
@@ -13,6 +14,7 @@ public final class org/dexpace/sdk/transport/jdkhttp/JdkHttpTransport$Builder :
 	public synthetic fun build ()Ljava/lang/Object;
 	public fun build ()Lorg/dexpace/sdk/transport/jdkhttp/JdkHttpTransport;
 	public final fun connectTimeout (Ljava/time/Duration;)Lorg/dexpace/sdk/transport/jdkhttp/JdkHttpTransport$Builder;
+	public final fun droppedHeaderLogging (Lorg/dexpace/sdk/core/instrumentation/DroppedHeaderLogging;)Lorg/dexpace/sdk/transport/jdkhttp/JdkHttpTransport$Builder;
 	public final fun followRedirects (Z)Lorg/dexpace/sdk/transport/jdkhttp/JdkHttpTransport$Builder;
 	public final fun httpVersion (Lorg/dexpace/sdk/transport/jdkhttp/JdkHttpTransport$HttpVersion;)Lorg/dexpace/sdk/transport/jdkhttp/JdkHttpTransport$Builder;
 	public final fun proxy (Lorg/dexpace/sdk/core/util/ProxyOptions;)Lorg/dexpace/sdk/transport/jdkhttp/JdkHttpTransport$Builder;
@@ -23,7 +25,8 @@ public final class org/dexpace/sdk/transport/jdkhttp/JdkHttpTransport$Companion
 	public final fun builder ()Lorg/dexpace/sdk/transport/jdkhttp/JdkHttpTransport$Builder;
 	public final fun create (Ljava/net/http/HttpClient;)Lorg/dexpace/sdk/transport/jdkhttp/JdkHttpTransport;
 	public final fun create (Ljava/net/http/HttpClient;Ljava/time/Duration;)Lorg/dexpace/sdk/transport/jdkhttp/JdkHttpTransport;
-	public static synthetic fun create$default (Lorg/dexpace/sdk/transport/jdkhttp/JdkHttpTransport$Companion;Ljava/net/http/HttpClient;Ljava/time/Duration;ILjava/lang/Object;)Lorg/dexpace/sdk/transport/jdkhttp/JdkHttpTransport;
+	public final fun create (Ljava/net/http/HttpClient;Ljava/time/Duration;Lorg/dexpace/sdk/core/instrumentation/DroppedHeaderLogging;)Lorg/dexpace/sdk/transport/jdkhttp/JdkHttpTransport;
+	public static synthetic fun create$default (Lorg/dexpace/sdk/transport/jdkhttp/JdkHttpTransport$Companion;Ljava/net/http/HttpClient;Ljava/time/Duration;Lorg/dexpace/sdk/core/instrumentation/DroppedHeaderLogging;ILjava/lang/Object;)Lorg/dexpace/sdk/transport/jdkhttp/JdkHttpTransport;
 }
 
 public final class org/dexpace/sdk/transport/jdkhttp/JdkHttpTransport$HttpVersion : java/lang/Enum {
diff --git a/sdk-transport-jdkhttp/src/main/kotlin/org/dexpace/sdk/transport/jdkhttp/JdkHttpTransport.kt b/sdk-transport-jdkhttp/src/main/kotlin/org/dexpace/sdk/transport/jdkhttp/JdkHttpTransport.kt
index 85c60c04..6abaa7ad 100644
--- a/sdk-transport-jdkhttp/src/main/kotlin/org/dexpace/sdk/transport/jdkhttp/JdkHttpTransport.kt
+++ b/sdk-transport-jdkhttp/src/main/kotlin/org/dexpace/sdk/transport/jdkhttp/JdkHttpTransport.kt
@@ -12,6 +12,7 @@ import org.dexpace.sdk.core.client.HttpClient
 import org.dexpace.sdk.core.http.request.Request
 import org.dexpace.sdk.core.http.response.Response
 import org.dexpace.sdk.core.instrumentation.ClientLogger
+import org.dexpace.sdk.core.instrumentation.DroppedHeaderLogging
 import org.dexpace.sdk.core.util.ProxyOptions
 import org.dexpace.sdk.transport.jdkhttp.internal.RequestAdapter
 import org.dexpace.sdk.transport.jdkhttp.internal.ResponseAdapter
@@ -84,9 +85,10 @@ public class JdkHttpTransport private constructor(
      * left untouched.
      */
     private val ownedExecutor: ExecutorService?,
+    droppedHeaderLogging: DroppedHeaderLogging,
 ) : HttpClient, AsyncHttpClient {
     private val log: ClientLogger = ClientLogger("org.dexpace.sdk.transport.jdkhttp.JdkHttpTransport")
-    private val requestAdapter: RequestAdapter = RequestAdapter(log)
+    private val requestAdapter: RequestAdapter = RequestAdapter(log, droppedHeaderLogging)
     private val responseAdapter: ResponseAdapter = ResponseAdapter(log)
 
     /**
@@ -241,13 +243,24 @@ public class JdkHttpTransport private constructor(
          * [responseTimeout], when non-`null`, is applied to every outgoing request via
          * [java.net.http.HttpRequest.Builder.timeout] — the JDK client does not expose a
          * global response-timeout knob. Defaults to `null` (no per-request timeout).
+         *
+         * @param droppedHeaderLogging how the transport logs a model-valid header the JDK client
+         *   cannot encode and therefore drops. Defaults to [DroppedHeaderLogging.ONCE_PER_HEADER].
          */
         @JvmStatic
         @JvmOverloads
         public fun create(
             client: java.net.http.HttpClient,
             responseTimeout: Duration? = null,
-        ): JdkHttpTransport = JdkHttpTransport(client, responseTimeout, owned = false, ownedExecutor = null)
+            droppedHeaderLogging: DroppedHeaderLogging = DroppedHeaderLogging.ONCE_PER_HEADER,
+        ): JdkHttpTransport =
+            JdkHttpTransport(
+                client,
+                responseTimeout,
+                owned = false,
+                ownedExecutor = null,
+                droppedHeaderLogging = droppedHeaderLogging,
+            )
 
         /** Returns a fresh [Builder] for SDK-managed [java.net.http.HttpClient] construction. */
         @JvmStatic
@@ -298,6 +311,7 @@ public class JdkHttpTransport private constructor(
         private var proxy: ProxyOptions? = null
         private var followRedirects: Boolean = false
         private var httpVersion: HttpVersion = HttpVersion.HTTP_2
+        private var droppedHeaderLogging: DroppedHeaderLogging = DroppedHeaderLogging.ONCE_PER_HEADER
 
         /** Sets the connect timeout (TCP handshake + TLS handshake). */
         public fun connectTimeout(d: Duration): Builder =
@@ -342,6 +356,17 @@ public class JdkHttpTransport private constructor(
                 this.httpVersion = v
             }
 
+        /**
+         * How the transport logs a model-valid header the JDK client cannot encode and therefore
+         * drops (e.g. a name with an interior space). Defaults to
+         * [DroppedHeaderLogging.ONCE_PER_HEADER] — one `WARN` the first time each such header is
+         * seen, the rest at verbose.
+         */
+        public fun droppedHeaderLogging(policy: DroppedHeaderLogging): Builder =
+            apply {
+                this.droppedHeaderLogging = policy
+            }
+
         /**
          * Builds a [JdkHttpTransport]. The underlying [java.net.http.HttpClient] is created
          * with the knobs configured above; unconfigured knobs fall through to the JDK's
@@ -371,7 +396,13 @@ public class JdkHttpTransport private constructor(
             // executor when one isn't supplied via `Builder.executor`. When the Builder
             // exposes an executor knob in the future, capture it here and pass it through
             // to the [JdkHttpTransport] constructor so [close] can shut it down.
-            return JdkHttpTransport(clientBuilder.build(), responseTimeout, owned = true, ownedExecutor = null)
+            return JdkHttpTransport(
+                clientBuilder.build(),
+                responseTimeout,
+                owned = true,
+                ownedExecutor = null,
+                droppedHeaderLogging = droppedHeaderLogging,
+            )
         }
 
         /**
diff --git a/sdk-transport-jdkhttp/src/main/kotlin/org/dexpace/sdk/transport/jdkhttp/internal/RequestAdapter.kt b/sdk-transport-jdkhttp/src/main/kotlin/org/dexpace/sdk/transport/jdkhttp/internal/RequestAdapter.kt
index 98c4bd38..97fef2ed 100644
--- a/sdk-transport-jdkhttp/src/main/kotlin/org/dexpace/sdk/transport/jdkhttp/internal/RequestAdapter.kt
+++ b/sdk-transport-jdkhttp/src/main/kotlin/org/dexpace/sdk/transport/jdkhttp/internal/RequestAdapter.kt
@@ -9,8 +9,12 @@ package org.dexpace.sdk.transport.jdkhttp.internal
 
 import org.dexpace.sdk.core.http.request.Method
 import org.dexpace.sdk.core.instrumentation.ClientLogger
+import org.dexpace.sdk.core.instrumentation.DroppedHeaderLogging
+import org.dexpace.sdk.core.instrumentation.LogLevel
 import java.net.http.HttpRequest
 import java.time.Duration
+import java.util.Locale
+import java.util.concurrent.ConcurrentHashMap
 import org.dexpace.sdk.core.http.request.Request as SdkRequest
 
 /**
@@ -29,12 +33,40 @@ import org.dexpace.sdk.core.http.request.Request as SdkRequest
  *     when non-null. JDK 11+ enforces the timeout from connect through response-headers
  *     receipt.
  *
- * Stateless — the [logger] is the only field, used to attribute the DEBUG log naming dropped
- * headers to the transport.
+ * Carries the [logger] (to attribute the dropped-header log to the transport) and the
+ * [droppedHeaderLogging] policy with its per-header WARN-dedup state; otherwise stateless and safe
+ * to share across requests and threads.
  */
 internal class RequestAdapter(
     private val logger: ClientLogger,
+    private val droppedHeaderLogging: DroppedHeaderLogging = DroppedHeaderLogging.ONCE_PER_HEADER,
 ) {
+    // Header names already WARN-logged under ONCE_PER_HEADER, keyed by lower-cased name so a re-set
+    // of the same header (any casing) warns at most once per transport. Thread-safe: an async
+    // transport adapts requests on many threads.
+    private val warnedDroppedHeaders: MutableSet<String> = ConcurrentHashMap.newKeySet()
+
+    /**
+     * Whether this drop of [rawName] should log at `WARN` (else verbose), per [droppedHeaderLogging].
+     * Internal for direct policy testing: the transport test modules run on `slf4j-nop` and cannot
+     * capture emitted log output, so the dedup decision is asserted here rather than through logs.
+     *
+     * Callers MUST gate this behind `logger.canLog(LogLevel.WARNING)` (see [attachHeaders]): under
+     * `ONCE_PER_HEADER` the call has the side effect of recording [rawName], so spending it while
+     * `WARN` is disabled would consume the one-shot without emitting anything and hide the drop if
+     * the level is later raised. The `ONCE_PER_HEADER` set is bounded at [MAX_WARNED_DROPPED_HEADERS]
+     * entries so a caller synthesising unbounded distinct non-token names cannot grow it without
+     * limit on a transport-lifetime adapter; past the bound, further first-drops log at verbose.
+     */
+    internal fun shouldWarnForDroppedHeader(rawName: String): Boolean =
+        when (droppedHeaderLogging) {
+            DroppedHeaderLogging.PER_OCCURRENCE -> true
+            DroppedHeaderLogging.ONCE_PER_HEADER ->
+                warnedDroppedHeaders.size < MAX_WARNED_DROPPED_HEADERS &&
+                    warnedDroppedHeaders.add(rawName.lowercase(Locale.US))
+            DroppedHeaderLogging.VERBOSE_ONLY -> false
+        }
+
     /**
      * Adapts [request] into a JDK [HttpRequest]. [responseTimeout] is applied per-request via
      * `HttpRequest.Builder.timeout(...)` when non-null.
@@ -104,13 +136,12 @@ internal class RequestAdapter(
      * `IllegalArgumentException` escape [adapt] (and therefore `execute`, declared
      * `@Throws(IOException)`) where a caller's `catch(IOException)` would not observe it.
      *
-     * Upstream `Headers.Builder` validation closes the request/header-splitting surface
-     * (control-character names, and control-character values bar horizontal tab, are rejected
-     * before they reach here), but it does not mirror the JDK's full field-name/value grammar.
-     * The `IllegalArgumentException` caught
-     * here is therefore the JDK refusing either a name in its restricted set or a model-valid
-     * name/value it nonetheless rejects (e.g. a non-token / non-ASCII byte the SDK deliberately
-     * permits) — not a control-character splitting vector, which never gets this far.
+     * Upstream `Headers.Builder` validation restricts names and values to printable ASCII (control
+     * characters and non-ASCII bytes are rejected at construction), closing the request/header-
+     * splitting surface, but it does not mirror the JDK's full field-name/value token grammar. The
+     * `IllegalArgumentException` caught here is therefore the JDK refusing either a name in its
+     * restricted set or a model-valid but non-token name/value (e.g. one carrying an interior
+     * space) — not a control-character splitting vector, which never gets this far.
      */
     private fun attachHeaders(
         builder: HttpRequest.Builder,
@@ -128,17 +159,32 @@ internal class RequestAdapter(
                 try {
                     builder.header(rawName, value)
                 } catch (e: IllegalArgumentException) {
-                    // Warn (not verbose): this is a header the caller explicitly set being silently
-                    // dropped because this transport cannot encode it — surfaced by default so the
-                    // loss is visible. Restricted-header drops above stay at verbose (expected, the
-                    // JDK recomputes or forbids them), as does the inbound response-header drop.
-                    logger.atWarning()
+                    // The model already rejects at construction the names/values no transport can
+                    // encode (control + non-ASCII), so this residual drop is a model-valid but
+                    // non-token header only this transport refuses. How loudly it surfaces is the
+                    // caller's DroppedHeaderLogging choice; the default warns once per header name and
+                    // logs the rest at verbose so a hot path re-setting it does not flood the log.
+                    // Gate the dedup on canLog(WARNING) so its one-shot side effect is spent only when
+                    // WARN can actually be emitted (see shouldWarnForDroppedHeader).
+                    val warn = logger.canLog(LogLevel.WARNING) && shouldWarnForDroppedHeader(rawName)
+                    val event = if (warn) logger.atWarning() else logger.atVerbose()
+                    event
                         .event("transport.jdkhttp.header.rejected")
                         .field("name", rawName)
                         .cause(e)
-                        .log("JDK rejected header name/value; dropping before dispatch")
+                        .log("JDK rejected header name/value it cannot encode; dropping before dispatch")
                 }
             }
         }
     }
+
+    private companion object {
+        /**
+         * Upper bound on the `ONCE_PER_HEADER` dedup set. The adapter lives for the transport's
+         * lifetime, so an unbounded set would leak one entry per distinct dropped header name; a
+         * caller deriving names from request data could grow it without limit. Past this many
+         * distinct names, further first-drops fall through to verbose rather than being recorded.
+         */
+        private const val MAX_WARNED_DROPPED_HEADERS: Int = 1024
+    }
 }
diff --git a/sdk-transport-jdkhttp/src/main/kotlin/org/dexpace/sdk/transport/jdkhttp/internal/ResponseAdapter.kt b/sdk-transport-jdkhttp/src/main/kotlin/org/dexpace/sdk/transport/jdkhttp/internal/ResponseAdapter.kt
index 4c07b632..7f9aba4c 100644
--- a/sdk-transport-jdkhttp/src/main/kotlin/org/dexpace/sdk/transport/jdkhttp/internal/ResponseAdapter.kt
+++ b/sdk-transport-jdkhttp/src/main/kotlin/org/dexpace/sdk/transport/jdkhttp/internal/ResponseAdapter.kt
@@ -98,14 +98,16 @@ internal class ResponseAdapter(
     /**
      * Copies one inbound (response) header into [headersBuilder].
      *
-     * The model layer's `add` validation is an **outbound** injection guard — it stops an SDK
-     * caller from putting a CR/LF or other control character into a *request* that is then
-     * serialised to a server. A response has already been received; the JDK client parses response
-     * headers leniently and can surface a control byte (notably over HTTP/2), so a strict
-     * re-validation here would let one odd server header throw out of `add` and, via the outer
-     * catch, fail the entire response. Catch that rejection and drop just the offending header —
-     * the body and the rest of the headers still reach the caller. The header name is logged (not
-     * the value, which may be sensitive) at verbose, mirroring the request adapter's drop-and-log.
+     * A response has already been received, so the outbound field-value grammar does not apply:
+     * RFC 7230 permits obs-text (`0x80`+) in a field-value, and the JDK client parses response
+     * headers leniently and surfaces such bytes (e.g. a Latin-1 `Content-Disposition` filename).
+     * Copying through the strict [Headers.Builder.add] would reject those and silently strip
+     * legitimate headers, so this uses [Headers.Builder.addUnsafeNonAscii], which preserves
+     * non-ASCII bytes while still rejecting control characters (a CR/LF/NUL has no place in a
+     * parsed value). A genuinely malformed header — a control byte in the value, or a control/non-
+     * ASCII byte in the name — still throws; catch that and drop just the offending header so the
+     * body and the rest of the headers still reach the caller. The header name is logged (not the
+     * value, which may be sensitive) at verbose, mirroring the request adapter's drop-and-log.
      */
     private fun addInboundHeader(
         headersBuilder: Headers.Builder,
@@ -113,7 +115,7 @@ internal class ResponseAdapter(
         value: String,
     ) {
         try {
-            headersBuilder.add(name, value)
+            headersBuilder.addUnsafeNonAscii(name, value)
         } catch (e: IllegalArgumentException) {
             logger.atVerbose()
                 .event("transport.jdkhttp.response.header.dropped")
diff --git a/sdk-transport-jdkhttp/src/test/kotlin/org/dexpace/sdk/transport/jdkhttp/JdkHttpTransportTest.kt b/sdk-transport-jdkhttp/src/test/kotlin/org/dexpace/sdk/transport/jdkhttp/JdkHttpTransportTest.kt
index 03121bd5..15bfa6fe 100644
--- a/sdk-transport-jdkhttp/src/test/kotlin/org/dexpace/sdk/transport/jdkhttp/JdkHttpTransportTest.kt
+++ b/sdk-transport-jdkhttp/src/test/kotlin/org/dexpace/sdk/transport/jdkhttp/JdkHttpTransportTest.kt
@@ -17,6 +17,7 @@ import org.dexpace.sdk.core.http.common.Protocol
 import org.dexpace.sdk.core.http.request.Method
 import org.dexpace.sdk.core.http.request.Request
 import org.dexpace.sdk.core.http.request.RequestBody
+import org.dexpace.sdk.core.instrumentation.DroppedHeaderLogging
 import org.dexpace.sdk.core.io.BufferedSink
 import org.dexpace.sdk.core.io.Io
 import org.dexpace.sdk.core.util.ProxyOptions
@@ -292,6 +293,29 @@ class JdkHttpTransportTest {
         assertTrue(xCustomValues.containsAll(listOf("alpha", "beta")), "expected both custom values in $xCustomValues")
     }
 
+    @Test
+    fun `inboundResponseHeaderWithNonAsciiValueIsPreserved`() {
+        // RFC 7230 permits obs-text (0x80+) in a field-value — e.g. a Latin-1 Content-Disposition
+        // filename — and the JDK client decodes response header bytes as ISO-8859-1, surfacing them.
+        // The response adapter must preserve the header, not silently drop it (the strict outbound
+        // grammar would reject it; the inbound path relaxes only the ASCII restriction).
+        val eAcute = 233.toChar() // 'é' (U+00E9)
+        server.enqueue(
+            MockResponse.Builder()
+                .code(200)
+                // addHeaderLenient bypasses MockWebServer's own strict value grammar so the raw
+                // obs-text byte reaches the wire, exactly as a real server would send it.
+                .addHeaderLenient("Content-Disposition", "attachment; filename=\"caf" + eAcute + ".pdf\"")
+                .build(),
+        )
+        val request = Request.builder().method(Method.GET).url(server.url("/obs-text").toUrl()).build()
+        transport.execute(request).use { response ->
+            val value = response.headers.get("Content-Disposition")
+            assertNotNull(value, "obs-text response header must be preserved, not dropped")
+            assertTrue(value.any { it.code >= 0x80 }, "the non-ASCII byte must survive: $value")
+        }
+    }
+
     // -------- restricted headers --------
 
     @Test
@@ -330,18 +354,19 @@ class JdkHttpTransportTest {
 
     @Test
     fun `headerRejectedByJdkIsDroppedNotThrown`() {
-        // The SDK model layer permits a non-ASCII header name (it rejects only control characters),
-        // but the JDK's field-name grammar rejects a non-token byte. The adapter must drop it rather
-        // than let the unchecked IllegalArgumentException escape execute()'s @Throws(IOException)
-        // contract — the same drop-and-log path the OkHttp adapter now mirrors. Built with toChar()
-        // so the offending byte is unambiguous in source.
-        val oUmlaut = 246.toChar() // 'o' with diaeresis (U+00F6): not an RFC 7230 token char
+        // The SDK model layer permits any printable-ASCII header name (it rejects only control and
+        // non-ASCII bytes), but the JDK's field-name grammar rejects a non-token byte. The adapter
+        // must drop it rather than let the unchecked IllegalArgumentException escape execute()'s
+        // @Throws(IOException) contract — the same drop-and-log path the OkHttp adapter mirrors. A
+        // space (0x20) is a model-valid, non-token name byte; built with toChar() so it is
+        // unambiguous in source.
+        val space = 32.toChar() // 0x20: printable ASCII the model permits, but not an RFC 7230 token char
         server.enqueue(MockResponse.Builder().code(200).body("ok").build())
         val request =
             Request.builder()
                 .method(Method.GET)
                 .url(server.url("/non-token-name").toUrl())
-                .addHeader("X-Uni" + oUmlaut + "code", "plain")
+                .addHeader("X-Uni" + space + "code", "plain")
                 .addHeader("X-Pass-Through", "kept")
                 .build()
         // execute must NOT throw; the rejected header is simply absent on the wire.
@@ -349,22 +374,52 @@ class JdkHttpTransportTest {
             assertEquals(200, response.status.code)
         }
         val recorded = server.takeRequest()
-        assertNull(recorded.headers["X-Uni" + oUmlaut + "code"], "non-token name must be dropped")
+        assertNull(recorded.headers["X-Uni" + space + "code"], "non-token name must be dropped")
+        assertEquals("kept", recorded.headers["X-Pass-Through"])
+    }
+
+    @Test
+    fun `droppedHeaderLoggingPolicyIsWiredThroughTheBuilder`() {
+        // Wiring smoke test: a transport built with a non-default DroppedHeaderLogging still adapts
+        // the request, drops the non-token header, and dispatches. The log *level* is not asserted
+        // (tests run on slf4j-nop); the policy decision itself is unit-tested in
+        // RequestAdapterDroppedHeaderLoggingTest.
+        val space = 32.toChar() // 0x20: model-valid printable ASCII, not an RFC 7230 token char
+        server.enqueue(MockResponse.Builder().code(200).body("ok").build())
+        val request =
+            Request.builder()
+                .method(Method.GET)
+                .url(server.url("/policy-wiring").toUrl())
+                .addHeader("X-Uni" + space + "code", "plain")
+                .addHeader("X-Pass-Through", "kept")
+                .build()
+        JdkHttpTransport.builder()
+            .httpVersion(JdkHttpTransport.HttpVersion.HTTP_1_1)
+            .droppedHeaderLogging(DroppedHeaderLogging.VERBOSE_ONLY)
+            .build()
+            .use { policyTransport ->
+                policyTransport.execute(request).use { response ->
+                    assertEquals(200, response.status.code)
+                }
+            }
+        val recorded = server.takeRequest()
+        assertNull(recorded.headers["X-Uni" + space + "code"], "non-token name must be dropped under any policy")
         assertEquals("kept", recorded.headers["X-Pass-Through"])
     }
 
     @Test
     fun `headerRejectedByJdkIsDroppedNotThrownAsync`() {
-        // The adapter runs on the async path too. A header the JDK rejects must be dropped so the
-        // future completes normally (not exceptionally) and the request still dispatches —
-        // mirroring the sync drop-not-throw path and the OkHttp transport's async coverage.
-        val oUmlaut = 246.toChar() // 'o' with diaeresis (U+00F6): not an RFC 7230 token char
+        // The adapter runs on the async path too. A header the JDK rejects (here a non-token name
+        // carrying an interior space) must be dropped so the future completes normally (not
+        // exceptionally) and the request still dispatches — mirroring the sync drop-not-throw path
+        // and the OkHttp transport's async coverage.
+        val space = 32.toChar() // 0x20: model-valid printable ASCII, not an RFC 7230 token char
         server.enqueue(MockResponse.Builder().code(200).body("ok").build())
         val request =
             Request.builder()
                 .method(Method.GET)
                 .url(server.url("/non-token-name-async").toUrl())
-                .addHeader("X-Uni" + oUmlaut + "code", "plain")
+                .addHeader("X-Uni" + space + "code", "plain")
                 .addHeader("X-Pass-Through", "kept")
                 .build()
         val response = transport.executeAsync(request).get(5, TimeUnit.SECONDS)
@@ -372,7 +427,7 @@ class JdkHttpTransportTest {
             assertEquals(200, it.status.code)
         }
         val recorded = server.takeRequest()
-        assertNull(recorded.headers["X-Uni" + oUmlaut + "code"], "non-token name must be dropped")
+        assertNull(recorded.headers["X-Uni" + space + "code"], "non-token name must be dropped")
         assertEquals("kept", recorded.headers["X-Pass-Through"])
     }
 
diff --git a/sdk-transport-jdkhttp/src/test/kotlin/org/dexpace/sdk/transport/jdkhttp/internal/RequestAdapterDroppedHeaderLoggingTest.kt b/sdk-transport-jdkhttp/src/test/kotlin/org/dexpace/sdk/transport/jdkhttp/internal/RequestAdapterDroppedHeaderLoggingTest.kt
new file mode 100644
index 00000000..b77d3581
--- /dev/null
+++ b/sdk-transport-jdkhttp/src/test/kotlin/org/dexpace/sdk/transport/jdkhttp/internal/RequestAdapterDroppedHeaderLoggingTest.kt
@@ -0,0 +1,56 @@
+/*
+ * Copyright (c) 2026 dexpace and Omar Aljarrah
+ *
+ * Licensed under the MIT License. See LICENSE in the project root.
+ * SPDX-License-Identifier: MIT
+ */
+
+package org.dexpace.sdk.transport.jdkhttp.internal
+
+import org.dexpace.sdk.core.instrumentation.ClientLogger
+import org.dexpace.sdk.core.instrumentation.DroppedHeaderLogging
+import kotlin.test.Test
+import kotlin.test.assertFalse
+import kotlin.test.assertTrue
+
+/**
+ * Unit-tests the [DroppedHeaderLogging] decision on [RequestAdapter]. The transport test modules run
+ * on `slf4j-nop`, so emitted log output cannot be captured; the dedup decision that drives the log
+ * level is asserted directly through the internal `shouldWarnForDroppedHeader`.
+ */
+class RequestAdapterDroppedHeaderLoggingTest {
+    private fun adapter(policy: DroppedHeaderLogging): RequestAdapter =
+        RequestAdapter(ClientLogger("test.jdkhttp.RequestAdapter"), policy)
+
+    @Test
+    fun `PER_OCCURRENCE warns on every drop`() {
+        val adapter = adapter(DroppedHeaderLogging.PER_OCCURRENCE)
+        assertTrue(adapter.shouldWarnForDroppedHeader("X-Foo"))
+        assertTrue(adapter.shouldWarnForDroppedHeader("X-Foo"), "every drop of the same header still warns")
+        assertTrue(adapter.shouldWarnForDroppedHeader("X-Bar"))
+    }
+
+    @Test
+    fun `ONCE_PER_HEADER warns on the first drop per name then goes verbose, case-insensitively`() {
+        val adapter = adapter(DroppedHeaderLogging.ONCE_PER_HEADER)
+        assertTrue(adapter.shouldWarnForDroppedHeader("X-Foo"), "first drop of a header warns")
+        assertFalse(adapter.shouldWarnForDroppedHeader("X-Foo"), "repeat drop of the same header is verbose")
+        assertFalse(adapter.shouldWarnForDroppedHeader("x-foo"), "dedup is case-insensitive")
+        assertTrue(adapter.shouldWarnForDroppedHeader("X-Bar"), "a different header still warns once")
+        assertFalse(adapter.shouldWarnForDroppedHeader("X-Bar"))
+    }
+
+    @Test
+    fun `VERBOSE_ONLY never warns`() {
+        val adapter = adapter(DroppedHeaderLogging.VERBOSE_ONLY)
+        assertFalse(adapter.shouldWarnForDroppedHeader("X-Foo"))
+        assertFalse(adapter.shouldWarnForDroppedHeader("X-Foo"))
+    }
+
+    @Test
+    fun `the default policy is ONCE_PER_HEADER`() {
+        val adapter = RequestAdapter(ClientLogger("test.jdkhttp.RequestAdapter"))
+        assertTrue(adapter.shouldWarnForDroppedHeader("X-Foo"), "first drop warns under the default")
+        assertFalse(adapter.shouldWarnForDroppedHeader("X-Foo"), "default dedups after the first warn")
+    }
+}
diff --git a/sdk-transport-okhttp/api/sdk-transport-okhttp.api b/sdk-transport-okhttp/api/sdk-transport-okhttp.api
index 99ca3d97..529de89c 100644
--- a/sdk-transport-okhttp/api/sdk-transport-okhttp.api
+++ b/sdk-transport-okhttp/api/sdk-transport-okhttp.api
@@ -1,9 +1,10 @@
 public final class org/dexpace/sdk/transport/okhttp/OkHttpTransport : org/dexpace/sdk/core/client/AsyncHttpClient, org/dexpace/sdk/core/client/HttpClient {
 	public static final field Companion Lorg/dexpace/sdk/transport/okhttp/OkHttpTransport$Companion;
-	public synthetic fun <init> (Lokhttp3/OkHttpClient;ZLkotlin/jvm/internal/DefaultConstructorMarker;)V
+	public synthetic fun <init> (Lokhttp3/OkHttpClient;ZLorg/dexpace/sdk/core/instrumentation/DroppedHeaderLogging;Lkotlin/jvm/internal/DefaultConstructorMarker;)V
 	public static final fun builder ()Lorg/dexpace/sdk/transport/okhttp/OkHttpTransport$Builder;
 	public fun close ()V
 	public static final fun create (Lokhttp3/OkHttpClient;)Lorg/dexpace/sdk/transport/okhttp/OkHttpTransport;
+	public static final fun create (Lokhttp3/OkHttpClient;Lorg/dexpace/sdk/core/instrumentation/DroppedHeaderLogging;)Lorg/dexpace/sdk/transport/okhttp/OkHttpTransport;
 	public fun execute (Lorg/dexpace/sdk/core/http/request/Request;)Lorg/dexpace/sdk/core/http/response/Response;
 	public fun executeAsync (Lorg/dexpace/sdk/core/http/request/Request;)Ljava/util/concurrent/CompletableFuture;
 }
@@ -13,6 +14,7 @@ public final class org/dexpace/sdk/transport/okhttp/OkHttpTransport$Builder : or
 	public fun build ()Lorg/dexpace/sdk/transport/okhttp/OkHttpTransport;
 	public final fun callTimeout (Ljava/time/Duration;)Lorg/dexpace/sdk/transport/okhttp/OkHttpTransport$Builder;
 	public final fun connectTimeout (Ljava/time/Duration;)Lorg/dexpace/sdk/transport/okhttp/OkHttpTransport$Builder;
+	public final fun droppedHeaderLogging (Lorg/dexpace/sdk/core/instrumentation/DroppedHeaderLogging;)Lorg/dexpace/sdk/transport/okhttp/OkHttpTransport$Builder;
 	public final fun followRedirects (Z)Lorg/dexpace/sdk/transport/okhttp/OkHttpTransport$Builder;
 	public final fun proxy (Lorg/dexpace/sdk/core/util/ProxyOptions;)Lorg/dexpace/sdk/transport/okhttp/OkHttpTransport$Builder;
 	public final fun readTimeout (Ljava/time/Duration;)Lorg/dexpace/sdk/transport/okhttp/OkHttpTransport$Builder;
@@ -22,5 +24,7 @@ public final class org/dexpace/sdk/transport/okhttp/OkHttpTransport$Builder : or
 public final class org/dexpace/sdk/transport/okhttp/OkHttpTransport$Companion {
 	public final fun builder ()Lorg/dexpace/sdk/transport/okhttp/OkHttpTransport$Builder;
 	public final fun create (Lokhttp3/OkHttpClient;)Lorg/dexpace/sdk/transport/okhttp/OkHttpTransport;
+	public final fun create (Lokhttp3/OkHttpClient;Lorg/dexpace/sdk/core/instrumentation/DroppedHeaderLogging;)Lorg/dexpace/sdk/transport/okhttp/OkHttpTransport;
+	public static synthetic fun create$default (Lorg/dexpace/sdk/transport/okhttp/OkHttpTransport$Companion;Lokhttp3/OkHttpClient;Lorg/dexpace/sdk/core/instrumentation/DroppedHeaderLogging;ILjava/lang/Object;)Lorg/dexpace/sdk/transport/okhttp/OkHttpTransport;
 }
 
diff --git a/sdk-transport-okhttp/src/main/kotlin/org/dexpace/sdk/transport/okhttp/OkHttpTransport.kt b/sdk-transport-okhttp/src/main/kotlin/org/dexpace/sdk/transport/okhttp/OkHttpTransport.kt
index 2cea0f82..8e99ba25 100644
--- a/sdk-transport-okhttp/src/main/kotlin/org/dexpace/sdk/transport/okhttp/OkHttpTransport.kt
+++ b/sdk-transport-okhttp/src/main/kotlin/org/dexpace/sdk/transport/okhttp/OkHttpTransport.kt
@@ -17,6 +17,7 @@ import org.dexpace.sdk.core.client.HttpClient
 import org.dexpace.sdk.core.http.request.Request
 import org.dexpace.sdk.core.http.response.Response
 import org.dexpace.sdk.core.instrumentation.ClientLogger
+import org.dexpace.sdk.core.instrumentation.DroppedHeaderLogging
 import org.dexpace.sdk.core.util.ProxyOptions
 import org.dexpace.sdk.transport.okhttp.internal.RequestAdapter
 import org.dexpace.sdk.transport.okhttp.internal.ResponseAdapter
@@ -67,8 +68,9 @@ public class OkHttpTransport private constructor(
      * decision in [close] per the SDK's ownership-aware lifecycle contract.
      */
     private val owned: Boolean,
+    droppedHeaderLogging: DroppedHeaderLogging,
 ) : HttpClient, AsyncHttpClient {
-    private val requestAdapter: RequestAdapter = RequestAdapter(log)
+    private val requestAdapter: RequestAdapter = RequestAdapter(log, droppedHeaderLogging)
     private val responseAdapter: ResponseAdapter = ResponseAdapter(log)
 
     /**
@@ -260,9 +262,16 @@ public class OkHttpTransport private constructor(
          * BYO factory: wrap a fully-configured [OkHttpClient]. The supplied client is used
          * verbatim — the SDK does not override `followRedirects`, timeouts, or interceptors,
          * and [close] will NOT shut down this client (the caller owns its lifecycle).
+         *
+         * @param droppedHeaderLogging how the transport logs a model-valid header OkHttp cannot
+         *   encode and therefore drops. Defaults to [DroppedHeaderLogging.ONCE_PER_HEADER].
          */
         @JvmStatic
-        public fun create(client: OkHttpClient): OkHttpTransport = OkHttpTransport(client, owned = false)
+        @JvmOverloads
+        public fun create(
+            client: OkHttpClient,
+            droppedHeaderLogging: DroppedHeaderLogging = DroppedHeaderLogging.ONCE_PER_HEADER,
+        ): OkHttpTransport = OkHttpTransport(client, owned = false, droppedHeaderLogging = droppedHeaderLogging)
 
         /** Returns a fresh [Builder] for SDK-managed [OkHttpClient] construction. */
         @JvmStatic
@@ -285,6 +294,7 @@ public class OkHttpTransport private constructor(
         private var callTimeout: Duration? = null
         private var proxy: ProxyOptions? = null
         private var followRedirects: Boolean = false
+        private var droppedHeaderLogging: DroppedHeaderLogging = DroppedHeaderLogging.ONCE_PER_HEADER
 
         /** Sets the connect timeout (TCP handshake + TLS handshake). */
         public fun connectTimeout(d: Duration): Builder =
@@ -327,6 +337,16 @@ public class OkHttpTransport private constructor(
                 this.followRedirects = enabled
             }
 
+        /**
+         * How the transport logs a model-valid header OkHttp cannot encode and therefore drops
+         * (e.g. a name with an interior space). Defaults to [DroppedHeaderLogging.ONCE_PER_HEADER]
+         * — one `WARN` the first time each such header is seen, the rest at verbose.
+         */
+        public fun droppedHeaderLogging(policy: DroppedHeaderLogging): Builder =
+            apply {
+                this.droppedHeaderLogging = policy
+            }
+
         /**
          * Builds an [OkHttpTransport]. The underlying [OkHttpClient] is created with the
          * knobs configured above; unconfigured knobs fall through to OkHttp's library
@@ -349,7 +369,7 @@ public class OkHttpTransport private constructor(
             // consume-guard. Disable it on SDK-managed clients only; BYO clients keep their
             // own configuration.
             okBuilder.retryOnConnectionFailure(false)
-            return OkHttpTransport(okBuilder.build(), owned = true)
+            return OkHttpTransport(okBuilder.build(), owned = true, droppedHeaderLogging = droppedHeaderLogging)
         }
 
         /**
diff --git a/sdk-transport-okhttp/src/main/kotlin/org/dexpace/sdk/transport/okhttp/internal/RequestAdapter.kt b/sdk-transport-okhttp/src/main/kotlin/org/dexpace/sdk/transport/okhttp/internal/RequestAdapter.kt
index 8727a987..ca96e771 100644
--- a/sdk-transport-okhttp/src/main/kotlin/org/dexpace/sdk/transport/okhttp/internal/RequestAdapter.kt
+++ b/sdk-transport-okhttp/src/main/kotlin/org/dexpace/sdk/transport/okhttp/internal/RequestAdapter.kt
@@ -12,6 +12,10 @@ import okhttp3.RequestBody
 import okhttp3.RequestBody.Companion.toRequestBody
 import org.dexpace.sdk.core.http.request.FileRequestBody
 import org.dexpace.sdk.core.instrumentation.ClientLogger
+import org.dexpace.sdk.core.instrumentation.DroppedHeaderLogging
+import org.dexpace.sdk.core.instrumentation.LogLevel
+import java.util.Locale
+import java.util.concurrent.ConcurrentHashMap
 import org.dexpace.sdk.core.http.request.Request as SdkRequest
 import org.dexpace.sdk.core.http.request.RequestBody as SdkRequestBody
 
@@ -37,12 +41,40 @@ import org.dexpace.sdk.core.http.request.RequestBody as SdkRequestBody
  *     request body.")` — cannot reach here: `Request.RequestBuilder.build` rejects a body on
  *     GET/HEAD/TRACE/CONNECT at construction, so `request.body` is always `null` for those methods.
  *
- * The adapter is stateless — the [logger] is the only field it carries so the DEBUG log
- * naming dropped headers attributes to the transport.
+ * The adapter carries the [logger] (so the dropped-header log attributes to the transport) and the
+ * [droppedHeaderLogging] policy with its per-header WARN-dedup state; it is otherwise stateless and
+ * safe to share across requests and threads.
  */
 internal class RequestAdapter(
     private val logger: ClientLogger,
+    private val droppedHeaderLogging: DroppedHeaderLogging = DroppedHeaderLogging.ONCE_PER_HEADER,
 ) {
+    // Header names already WARN-logged under ONCE_PER_HEADER, keyed by lower-cased name so a re-set
+    // of the same header (any casing) warns at most once per transport. Thread-safe: an async
+    // transport adapts requests on many threads.
+    private val warnedDroppedHeaders: MutableSet<String> = ConcurrentHashMap.newKeySet()
+
+    /**
+     * Whether this drop of [rawName] should log at `WARN` (else verbose), per [droppedHeaderLogging].
+     * Internal for direct policy testing: the transport test modules run on `slf4j-nop` and cannot
+     * capture emitted log output, so the dedup decision is asserted here rather than through logs.
+     *
+     * Callers MUST gate this behind `logger.canLog(LogLevel.WARNING)` (see [adapt]): under
+     * `ONCE_PER_HEADER` the call has the side effect of recording [rawName], so spending it while
+     * `WARN` is disabled would consume the one-shot without emitting anything and hide the drop if
+     * the level is later raised. The `ONCE_PER_HEADER` set is bounded at [MAX_WARNED_DROPPED_HEADERS]
+     * entries so a caller synthesising unbounded distinct non-token names cannot grow it without
+     * limit on a transport-lifetime adapter; past the bound, further first-drops log at verbose.
+     */
+    internal fun shouldWarnForDroppedHeader(rawName: String): Boolean =
+        when (droppedHeaderLogging) {
+            DroppedHeaderLogging.PER_OCCURRENCE -> true
+            DroppedHeaderLogging.ONCE_PER_HEADER ->
+                warnedDroppedHeaders.size < MAX_WARNED_DROPPED_HEADERS &&
+                    warnedDroppedHeaders.add(rawName.lowercase(Locale.US))
+            DroppedHeaderLogging.VERBOSE_ONLY -> false
+        }
+
     fun adapt(request: SdkRequest): Request {
         val builder = Request.Builder().url(request.url.toExternalForm())
         for ((rawName, values) in request.headers.entries()) {
@@ -54,28 +86,33 @@ internal class RequestAdapter(
                 continue
             }
             for (value in values) {
-                // Header names and values are validated upstream by Headers.Builder (names reject
-                // control characters; values reject control characters except horizontal tab),
-                // which closes the request/header-splitting surface. OkHttp is stricter still: it
-                // rejects any byte outside printable ASCII in a name (0x21-0x7e) or value (tab and
-                // 0x20-0x7e), so a model-valid non-ASCII (e.g. UTF-8) name or value — which the SDK
-                // deliberately permits — would make addHeader throw an unchecked
-                // IllegalArgumentException. Catch it and drop just that header, mirroring the JDK
-                // transport's attachHeaders, so the exception never escapes adapt (and therefore
-                // sync execute, declared @Throws(IOException)) as an unchecked failure a caller's
-                // catch(IOException) would miss.
+                // Header names and values are validated upstream by Headers.Builder to printable
+                // ASCII (control characters and non-ASCII bytes are rejected at construction), which
+                // closes the request/header-splitting surface. OkHttp is stricter still: it restricts
+                // a name to 0x21-0x7e (printable ASCII minus space — a superset of the RFC 7230 token
+                // grammar, which also excludes separators like '(' and '@') and a value to tab plus
+                // 0x20-0x7e, so a model-valid but non-token header — e.g. a name carrying an interior
+                // space (0x20) — makes addHeader throw an unchecked IllegalArgumentException. Catch it and drop just
+                // that header, mirroring the JDK transport's attachHeaders, so the exception never
+                // escapes adapt (and therefore sync execute, declared @Throws(IOException)) as an
+                // unchecked failure a caller's catch(IOException) would miss.
                 try {
                     builder.addHeader(rawName, value)
                 } catch (e: IllegalArgumentException) {
-                    // Warn (not verbose): this is a header the caller explicitly set being silently
-                    // dropped because this transport cannot encode it — surfaced by default so the
-                    // loss is visible. Restricted-header drops below stay at verbose (expected, the
-                    // transport recomputes them), as does the inbound response-header drop.
-                    logger.atWarning()
+                    // The model already rejects at construction the names/values no transport can
+                    // encode (control + non-ASCII), so this residual drop is a model-valid but
+                    // non-token header only this transport refuses. How loudly it surfaces is the
+                    // caller's DroppedHeaderLogging choice; the default warns once per header name and
+                    // logs the rest at verbose so a hot path re-setting it does not flood the log.
+                    // Gate the dedup on canLog(WARNING) so its one-shot side effect is spent only when
+                    // WARN can actually be emitted (see shouldWarnForDroppedHeader).
+                    val warn = logger.canLog(LogLevel.WARNING) && shouldWarnForDroppedHeader(rawName)
+                    val event = if (warn) logger.atWarning() else logger.atVerbose()
+                    event
                         .event("transport.okhttp.header.rejected")
                         .field("name", rawName)
                         .cause(e)
-                        .log("OkHttp rejected header name/value; dropping before dispatch")
+                        .log("OkHttp rejected header name/value it cannot encode; dropping before dispatch")
                 }
             }
         }
@@ -123,5 +160,13 @@ internal class RequestAdapter(
          * stateless, so a single instance is safe to reuse across requests and threads.
          */
         private val EMPTY_BODY: RequestBody = ByteArray(0).toRequestBody()
+
+        /**
+         * Upper bound on the `ONCE_PER_HEADER` dedup set. The adapter lives for the transport's
+         * lifetime, so an unbounded set would leak one entry per distinct dropped header name; a
+         * caller deriving names from request data could grow it without limit. Past this many
+         * distinct names, further first-drops fall through to verbose rather than being recorded.
+         */
+        private const val MAX_WARNED_DROPPED_HEADERS: Int = 1024
     }
 }
diff --git a/sdk-transport-okhttp/src/main/kotlin/org/dexpace/sdk/transport/okhttp/internal/ResponseAdapter.kt b/sdk-transport-okhttp/src/main/kotlin/org/dexpace/sdk/transport/okhttp/internal/ResponseAdapter.kt
index a13d153e..fc059643 100644
--- a/sdk-transport-okhttp/src/main/kotlin/org/dexpace/sdk/transport/okhttp/internal/ResponseAdapter.kt
+++ b/sdk-transport-okhttp/src/main/kotlin/org/dexpace/sdk/transport/okhttp/internal/ResponseAdapter.kt
@@ -92,14 +92,16 @@ internal class ResponseAdapter(
     /**
      * Copies one inbound (response) header into [headersBuilder].
      *
-     * The model layer's `add` validation is an **outbound** injection guard — it stops an SDK
-     * caller from putting a CR/LF or other control character into a *request* that is then
-     * serialised to a server. A response has already been received; OkHttp parses response headers
-     * leniently and can surface a control byte (notably over HTTP/2), so a strict re-validation
-     * here would let one odd server header throw out of `add` and, via the outer catch, fail the
-     * entire response. Catch that rejection and drop just the offending header — the body and the
-     * rest of the headers still reach the caller. The header name is logged (not the value, which
-     * may be sensitive) at verbose, mirroring the request adapter's drop-and-log.
+     * A response has already been received, so the outbound field-value grammar does not apply:
+     * RFC 7230 permits obs-text (`0x80`+) in a field-value, and OkHttp parses response headers
+     * leniently and surfaces such bytes (e.g. a Latin-1 `Content-Disposition` filename). Copying
+     * through the strict [Headers.Builder.add] would reject those and silently strip legitimate
+     * headers, so this uses [Headers.Builder.addUnsafeNonAscii], which preserves non-ASCII bytes
+     * while still rejecting control characters (a CR/LF/NUL has no place in a parsed value). A
+     * genuinely malformed header — a control byte in the value, or a control/non-ASCII byte in the
+     * name — still throws; catch that and drop just the offending header so the body and the rest of
+     * the headers still reach the caller. The header name is logged (not the value, which may be
+     * sensitive) at verbose, mirroring the request adapter's drop-and-log.
      */
     private fun addInboundHeader(
         headersBuilder: Headers.Builder,
@@ -107,7 +109,7 @@ internal class ResponseAdapter(
         value: String,
     ) {
         try {
-            headersBuilder.add(name, value)
+            headersBuilder.addUnsafeNonAscii(name, value)
         } catch (e: IllegalArgumentException) {
             logger.atVerbose()
                 .event("transport.okhttp.response.header.dropped")
diff --git a/sdk-transport-okhttp/src/test/kotlin/org/dexpace/sdk/transport/okhttp/OkHttpTransportTest.kt b/sdk-transport-okhttp/src/test/kotlin/org/dexpace/sdk/transport/okhttp/OkHttpTransportTest.kt
index bbfcccd1..a3f41922 100644
--- a/sdk-transport-okhttp/src/test/kotlin/org/dexpace/sdk/transport/okhttp/OkHttpTransportTest.kt
+++ b/sdk-transport-okhttp/src/test/kotlin/org/dexpace/sdk/transport/okhttp/OkHttpTransportTest.kt
@@ -22,6 +22,7 @@ import org.dexpace.sdk.core.http.request.Request
 import org.dexpace.sdk.core.http.request.RequestBody
 import org.dexpace.sdk.core.http.response.Response
 import org.dexpace.sdk.core.http.response.Status
+import org.dexpace.sdk.core.instrumentation.DroppedHeaderLogging
 import org.dexpace.sdk.core.io.Io
 import org.dexpace.sdk.core.util.ProxyOptions
 import org.dexpace.sdk.io.OkioIoProvider
@@ -300,6 +301,28 @@ class OkHttpTransportTest {
         assertTrue(xCustomValues.containsAll(listOf("alpha", "beta")), "expected both custom values in $xCustomValues")
     }
 
+    @Test
+    fun inboundResponseHeaderWithNonAsciiValueIsPreserved() {
+        // RFC 7230 permits obs-text (0x80+) in a field-value — e.g. a Latin-1 Content-Disposition
+        // filename. The response adapter must surface it, not silently drop the header (the strict
+        // outbound grammar would reject it; the inbound path relaxes only the ASCII restriction).
+        val eAcute = 233.toChar() // 'é' (U+00E9)
+        server.enqueue(
+            MockResponse.Builder()
+                .code(200)
+                // addHeaderLenient bypasses MockWebServer's own strict value grammar so the raw
+                // obs-text byte reaches the wire, exactly as a real server would send it.
+                .addHeaderLenient("Content-Disposition", "attachment; filename=\"caf" + eAcute + ".pdf\"")
+                .build(),
+        )
+        val request = Request.builder().method(Method.GET).url(server.url("/obs-text").toUrl()).build()
+        transport.execute(request).use { response ->
+            val value = response.headers.get("Content-Disposition")
+            assertNotNull(value, "obs-text response header must be preserved, not dropped")
+            assertTrue(value.any { it.code >= 0x80 }, "the non-ASCII byte must survive: $value")
+        }
+    }
+
     // -------- restricted headers --------
 
     @Test
@@ -331,42 +354,70 @@ class OkHttpTransportTest {
 
     @Test
     fun headerRejectedByOkHttpStricterRuleIsDroppedNotThrown() {
-        // The SDK model layer permits non-ASCII header names/values (it rejects only control
-        // characters), but OkHttp restricts both to printable ASCII and throws an unchecked
-        // IllegalArgumentException otherwise. The adapter must drop such a header — mirroring the
-        // JDK transport — so the exception never escapes execute()'s @Throws(IOException) contract.
-        // The offending byte is built with toChar() to keep it unambiguous in source.
-        val oUmlaut = 246.toChar() // 'o' with diaeresis (U+00F6): valid UTF-8, not printable ASCII
+        // The SDK model layer permits any printable-ASCII header name (it rejects only control and
+        // non-ASCII bytes), but OkHttp restricts a name to the RFC 7230 token bytes 0x21-0x7e and
+        // throws an unchecked IllegalArgumentException otherwise. The adapter must drop such a header
+        // — mirroring the JDK transport — so the exception never escapes execute()'s
+        // @Throws(IOException) contract. A space (0x20) is a model-valid, non-token name byte; built
+        // with toChar() to keep it unambiguous in source.
+        val space = 32.toChar() // 0x20: printable ASCII the model permits, but not an RFC 7230 token char
         server.enqueue(MockResponse.Builder().code(200).body("ok").build())
         val request =
             Request.builder()
                 .method(Method.GET)
-                .url(server.url("/non-ascii").toUrl())
-                .addHeader("X-Uni" + oUmlaut + "code", "plain") // non-ASCII name
-                .addHeader("X-Plain", "v" + oUmlaut + "lue") // non-ASCII value
+                .url(server.url("/non-token-name").toUrl())
+                .addHeader("X-Uni" + space + "code", "plain") // non-token name (interior space)
                 .addHeader("X-Pass-Through", "kept")
                 .build()
-        // execute must NOT throw; the rejected headers are simply absent on the wire.
+        // execute must NOT throw; the rejected header is simply absent on the wire.
         transport.execute(request).use { response ->
             assertEquals(200, response.status.code)
         }
         val recorded = server.takeRequest()
-        assertNull(recorded.headers["X-Uni" + oUmlaut + "code"], "non-ASCII name must be dropped")
-        assertNull(recorded.headers["X-Plain"], "header carrying a non-ASCII value must be dropped")
+        assertNull(recorded.headers["X-Uni" + space + "code"], "non-token name must be dropped")
+        assertEquals("kept", recorded.headers["X-Pass-Through"])
+    }
+
+    @Test
+    fun droppedHeaderLoggingPolicyIsWiredThroughTheBuilder() {
+        // Wiring smoke test: a transport built with a non-default DroppedHeaderLogging still adapts
+        // the request, drops the non-token header, and dispatches. The log *level* is not asserted
+        // (tests run on slf4j-nop); the policy decision itself is unit-tested in
+        // RequestAdapterDroppedHeaderLoggingTest.
+        val space = 32.toChar() // 0x20: model-valid printable ASCII, not an RFC 7230 token char
+        server.enqueue(MockResponse.Builder().code(200).body("ok").build())
+        val request =
+            Request.builder()
+                .method(Method.GET)
+                .url(server.url("/policy-wiring").toUrl())
+                .addHeader("X-Uni" + space + "code", "plain")
+                .addHeader("X-Pass-Through", "kept")
+                .build()
+        OkHttpTransport.builder()
+            .droppedHeaderLogging(DroppedHeaderLogging.VERBOSE_ONLY)
+            .build()
+            .use { policyTransport ->
+                policyTransport.execute(request).use { response ->
+                    assertEquals(200, response.status.code)
+                }
+            }
+        val recorded = server.takeRequest()
+        assertNull(recorded.headers["X-Uni" + space + "code"], "non-token name must be dropped under any policy")
         assertEquals("kept", recorded.headers["X-Pass-Through"])
     }
 
     @Test
     fun headerRejectedByOkHttpStricterRuleIsDroppedNotThrownAsync() {
-        // The adapter runs on the async path too. A header OkHttp rejects must be dropped so the
-        // future completes normally (not exceptionally) and the request still dispatches.
-        val oUmlaut = 246.toChar() // 'o' with diaeresis (U+00F6)
+        // The adapter runs on the async path too. A header OkHttp rejects (here a non-token name
+        // carrying an interior space) must be dropped so the future completes normally (not
+        // exceptionally) and the request still dispatches.
+        val space = 32.toChar() // 0x20: model-valid printable ASCII, not an RFC 7230 token char
         server.enqueue(MockResponse.Builder().code(200).body("ok").build())
         val request =
             Request.builder()
                 .method(Method.GET)
-                .url(server.url("/non-ascii-async").toUrl())
-                .addHeader("X-Uni" + oUmlaut + "code", "plain")
+                .url(server.url("/non-token-name-async").toUrl())
+                .addHeader("X-Uni" + space + "code", "plain")
                 .addHeader("X-Pass-Through", "kept")
                 .build()
         val response = transport.executeAsync(request).get(5, TimeUnit.SECONDS)
@@ -374,7 +425,7 @@ class OkHttpTransportTest {
             assertEquals(200, it.status.code)
         }
         val recorded = server.takeRequest()
-        assertNull(recorded.headers["X-Uni" + oUmlaut + "code"], "non-ASCII name must be dropped")
+        assertNull(recorded.headers["X-Uni" + space + "code"], "non-token name must be dropped")
         assertEquals("kept", recorded.headers["X-Pass-Through"])
     }
 
diff --git a/sdk-transport-okhttp/src/test/kotlin/org/dexpace/sdk/transport/okhttp/internal/RequestAdapterDroppedHeaderLoggingTest.kt b/sdk-transport-okhttp/src/test/kotlin/org/dexpace/sdk/transport/okhttp/internal/RequestAdapterDroppedHeaderLoggingTest.kt
new file mode 100644
index 00000000..dd90d344
--- /dev/null
+++ b/sdk-transport-okhttp/src/test/kotlin/org/dexpace/sdk/transport/okhttp/internal/RequestAdapterDroppedHeaderLoggingTest.kt
@@ -0,0 +1,56 @@
+/*
+ * Copyright (c) 2026 dexpace and Omar Aljarrah
+ *
+ * Licensed under the MIT License. See LICENSE in the project root.
+ * SPDX-License-Identifier: MIT
+ */
+
+package org.dexpace.sdk.transport.okhttp.internal
+
+import org.dexpace.sdk.core.instrumentation.ClientLogger
+import org.dexpace.sdk.core.instrumentation.DroppedHeaderLogging
+import kotlin.test.Test
+import kotlin.test.assertFalse
+import kotlin.test.assertTrue
+
+/**
+ * Unit-tests the [DroppedHeaderLogging] decision on [RequestAdapter]. The transport test modules run
+ * on `slf4j-nop`, so emitted log output cannot be captured; the dedup decision that drives the log
+ * level is asserted directly through the internal `shouldWarnForDroppedHeader`.
+ */
+class RequestAdapterDroppedHeaderLoggingTest {
+    private fun adapter(policy: DroppedHeaderLogging): RequestAdapter =
+        RequestAdapter(ClientLogger("test.okhttp.RequestAdapter"), policy)
+
+    @Test
+    fun `PER_OCCURRENCE warns on every drop`() {
+        val adapter = adapter(DroppedHeaderLogging.PER_OCCURRENCE)
+        assertTrue(adapter.shouldWarnForDroppedHeader("X-Foo"))
+        assertTrue(adapter.shouldWarnForDroppedHeader("X-Foo"), "every drop of the same header still warns")
+        assertTrue(adapter.shouldWarnForDroppedHeader("X-Bar"))
+    }
+
+    @Test
+    fun `ONCE_PER_HEADER warns on the first drop per name then goes verbose, case-insensitively`() {
+        val adapter = adapter(DroppedHeaderLogging.ONCE_PER_HEADER)
+        assertTrue(adapter.shouldWarnForDroppedHeader("X-Foo"), "first drop of a header warns")
+        assertFalse(adapter.shouldWarnForDroppedHeader("X-Foo"), "repeat drop of the same header is verbose")
+        assertFalse(adapter.shouldWarnForDroppedHeader("x-foo"), "dedup is case-insensitive")
+        assertTrue(adapter.shouldWarnForDroppedHeader("X-Bar"), "a different header still warns once")
+        assertFalse(adapter.shouldWarnForDroppedHeader("X-Bar"))
+    }
+
+    @Test
+    fun `VERBOSE_ONLY never warns`() {
+        val adapter = adapter(DroppedHeaderLogging.VERBOSE_ONLY)
+        assertFalse(adapter.shouldWarnForDroppedHeader("X-Foo"))
+        assertFalse(adapter.shouldWarnForDroppedHeader("X-Foo"))
+    }
+
+    @Test
+    fun `the default policy is ONCE_PER_HEADER`() {
+        val adapter = RequestAdapter(ClientLogger("test.okhttp.RequestAdapter"))
+        assertTrue(adapter.shouldWarnForDroppedHeader("X-Foo"), "first drop warns under the default")
+        assertFalse(adapter.shouldWarnForDroppedHeader("X-Foo"), "default dedups after the first warn")
+    }
+}