refactor: apply idiomatic-Kotlin simplifications across the SDK

Readability cleanups across sdk-core and the adapter modules, replacing
"Java-in-Kotlin" patterns with their idiomatic equivalents. Every change is
body-only or adds a private member: no public signature changes (apiCheck
unchanged), and the full gated build — tests, detekt, ktlint, binary
compatibility, the Kover floor, and the R8 shrink guard — passes.

Collection operators replace hand-rolled loops and accumulators:
- HttpPipelineBuilder / AsyncHttpPipelineBuilder: arrayOfNulls + index-fill +
  unchecked cast -> Array(size) { ordered[it] }, dropping two
  @Suppress("UNCHECKED_CAST") and an obsolete KDoc workaround note.
- DigestChallengeHandler.pickChallenge: a five-continue accumulation loop ->
  challenges.mapNotNull(::toCandidate), with the per-challenge validation moved
  into an extracted toCandidate() that keeps one early return per gate.
- DefaultRedirectStep (filter/forEach), UrlRedactor (mapTo), ResponsePipeline
  (fold), Annotations (filterIsInstance), LinkHeaderPaginationStrategy
  (sequence/mapNotNull/firstOrNull), TristateModule (forEachIndexed).

Control flow and casts:
- RetryStep: when (val readyState = prepareNextAttempt(...)) subject form;
  data object Proceed.
- JdkHttpTransport: exhaustive when over the two-constant HttpVersion enum
  (no else, so a future constant fails to compile).
- WriteAllInto: subject when (read).
- RequestAdapter: capture request.body into a local so it smart-casts, dropping
  the cross-module !!.
- AuthChallengeParser: long || punctuation chains -> private Set<Char> membership.
- PageNumberPaginationStrategy: try/catch(NumberFormatException) ->
  raw?.toIntOrNull() ?: fallback.
- LoggableResponseBody.contentLength: nested if -> a single Elvis fold.

Boilerplate de-duplicated via private helpers:
- MdcAwareExecutor: the capture/wrap pair repeated across seven overrides ->
  two MdcSnapshot.wrap helpers (capture stays at the call site).
- SpanLoggingExtensions: per-key MDC restore -> restoreMdc().
- TeeSink: the tap-budget clamp arithmetic -> tapAllowance().
- Io.installProvider: hand-thrown IllegalStateException -> check(...) with a
  lazily built message.

Idiom polish:
- MediaType: manual StringBuilder -> buildString (capacity hint kept).
- DispatchContext: string concatenation -> templates.
- Configuration.parseDuration: Character.toUpperCase -> Char.uppercaseChar().
- RequestRebuilder: drop @Suppress("UNUSED_VARIABLE") via catch (ignored: ...).
- ThrowOnHttpErrorStep: explicit return type on an anonymous mediaType() override.
- ForeignSinkAdapter / ForeignSourceAdapter: return Timeout.NONE inline.
- RestrictedHeaders (okhttp + jdkhttp): drop the no-op .lowercase() on
  already-lower-case literals.

Author: OmarAlJarrah

sdk-async-virtualthreads/src/main/kotlin/org/dexpace/sdk/async/virtualthreads/MdcAwareExecutor.kt

@@ -44,32 +44,36 @@ import java.util.concurrent.TimeUnit
  * [VirtualThreadAsyncHttpClient] for the `close()` path so shutdown semantics are unchanged.
  */
 internal class MdcAwareExecutor(private val delegate: ExecutorService) : ExecutorService by delegate {
+    private fun MdcSnapshot.wrap(command: Runnable): Runnable = Runnable { withMdc { command.run() } }
+
+    private fun <T> MdcSnapshot.wrap(task: Callable<T>): Callable<T> = Callable { withMdc { task.call() } }
+
     override fun execute(command: Runnable) {
         val snapshot = MdcSnapshot.capture()
-        delegate.execute { snapshot.withMdc { command.run() } }
+        delegate.execute(snapshot.wrap(command))
     }
 
     override fun <T : Any?> submit(task: Callable<T>): Future<T> {
         val snapshot = MdcSnapshot.capture()
-        return delegate.submit(Callable { snapshot.withMdc { task.call() } })
+        return delegate.submit(snapshot.wrap(task))
     }
 
     override fun submit(task: Runnable): Future<*> {
         val snapshot = MdcSnapshot.capture()
-        return delegate.submit { snapshot.withMdc { task.run() } }
+        return delegate.submit(snapshot.wrap(task))
     }
 
     override fun <T : Any?> submit(
         task: Runnable,
         result: T,
     ): Future<T> {
         val snapshot = MdcSnapshot.capture()
-        return delegate.submit({ snapshot.withMdc { task.run() } }, result)
+        return delegate.submit(snapshot.wrap(task), result)
     }
 
     override fun <T : Any?> invokeAll(tasks: MutableCollection<out Callable<T>>): MutableList<Future<T>> {
         val snapshot = MdcSnapshot.capture()
-        return delegate.invokeAll(tasks.map { task -> Callable { snapshot.withMdc { task.call() } } })
+        return delegate.invokeAll(tasks.map { task -> snapshot.wrap(task) })
     }
 
     override fun <T : Any?> invokeAll(
@@ -78,12 +82,12 @@ internal class MdcAwareExecutor(private val delegate: ExecutorService) : Executo
         unit: TimeUnit,
     ): MutableList<Future<T>> {
         val snapshot = MdcSnapshot.capture()
-        return delegate.invokeAll(tasks.map { task -> Callable { snapshot.withMdc { task.call() } } }, timeout, unit)
+        return delegate.invokeAll(tasks.map { task -> snapshot.wrap(task) }, timeout, unit)
     }
 
     override fun <T : Any?> invokeAny(tasks: MutableCollection<out Callable<T>>): T {
         val snapshot = MdcSnapshot.capture()
-        return delegate.invokeAny(tasks.map { task -> Callable { snapshot.withMdc { task.call() } } })
+        return delegate.invokeAny(tasks.map { task -> snapshot.wrap(task) })
     }
 
     override fun <T : Any?> invokeAny(
@@ -92,6 +96,6 @@ internal class MdcAwareExecutor(private val delegate: ExecutorService) : Executo
         unit: TimeUnit,
     ): T {
         val snapshot = MdcSnapshot.capture()
-        return delegate.invokeAny(tasks.map { task -> Callable { snapshot.withMdc { task.call() } } }, timeout, unit)
+        return delegate.invokeAny(tasks.map { task -> snapshot.wrap(task) }, timeout, unit)
     }
 }

sdk-core/src/main/kotlin/org/dexpace/sdk/core/config/Configuration.kt

@@ -202,7 +202,7 @@ public class Configuration internal constructor(
             // ISO-8601 path: `PT5S`, `P1D`, etc. Reject negative durations (e.g. `PT-5S`) for the
             // same reason the shorthand path does below — downstream consumers (Clock.sleep,
             // Futures.delay) assume a non-negative duration and throw on a negative one.
-            if (Character.toUpperCase(raw[0]) == 'P') {
+            if (raw[0].uppercaseChar() == 'P') {
                 return try {
                     val d = Duration.parse(raw)
                     if (d.isNegative) null else d

sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/auth/AuthChallengeParser.kt

@@ -290,15 +290,15 @@ public object AuthChallengeParser {
         }
     }
 
+    private val TOKEN_PUNCTUATION: Set<Char> = "!#$%&'*+-.^_`|~".toSet()
+
+    private val TOKEN68_PUNCTUATION: Set<Char> = "-._~+/".toSet()
+
     /** RFC 7230 token char: ALPHA / DIGIT / one of the punctuation set. */
     private fun isTokenChar(c: Char): Boolean =
-        (c in 'a'..'z') || (c in 'A'..'Z') || (c in '0'..'9') ||
-            c == '!' || c == '#' || c == '$' || c == '%' || c == '&' ||
-            c == '\'' || c == '*' || c == '+' || c == '-' || c == '.' ||
-            c == '^' || c == '_' || c == '`' || c == '|' || c == '~'
+        (c in 'a'..'z') || (c in 'A'..'Z') || (c in '0'..'9') || c in TOKEN_PUNCTUATION
 
     /** RFC 7235 token68 char (excluding the trailing "=" pad, handled separately). */
     private fun isToken68Char(c: Char): Boolean =
-        (c in 'a'..'z') || (c in 'A'..'Z') || (c in '0'..'9') ||
-            c == '-' || c == '.' || c == '_' || c == '~' || c == '+' || c == '/'
+        (c in 'a'..'z') || (c in 'A'..'Z') || (c in '0'..'9') || c in TOKEN68_PUNCTUATION
 }

sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/auth/DigestChallengeHandler.kt

@@ -96,35 +96,18 @@ public class DigestChallengeHandler
          * - it carries a `realm` and `nonce`.
          *
          * Ordering: we scan [preferredAlgorithms] first, then within that scan the
-         * challenges. This ensures SHA-256 wins over MD5 even when MD5 appears first
+         * candidates. This ensures SHA-256 wins over MD5 even when MD5 appears first
          * in the header.
          *
-         * Each `continue` skips a challenge that fails a specific validation gate
-         * (scheme, realm, nonce, qop, algorithm). Collapsing to a single composite
-         * predicate would obscure which validation rejected the candidate when
-         * debugging Digest interop.
+         * Per-challenge validation is delegated to [toCandidate]; this scan only applies
+         * the algorithm-priority ordering, independent of the order challenges arrived in.
          */
-        @Suppress("LoopWithTooManyJumpStatements")
         private fun pickChallenge(
             challenges: List<AuthenticateChallenge>,
         ): Pair<AuthenticateChallenge, DigestAlgorithm>? {
             // Find all challenges that match Digest with a satisfiable qop/realm/nonce
             // — we'll filter by algorithm preference below.
-            val candidates = ArrayList<Pair<AuthenticateChallenge, DigestAlgorithm>>(challenges.size)
-            for (challenge in challenges) {
-                if (!challenge.scheme.equals("Digest", ignoreCase = true)) continue
-                if (challenge.parameters["realm"] == null) continue
-                if (challenge.parameters["nonce"] == null) continue
-                if (!qopSupportsAuth(challenge.parameters["qop"])) continue
-                val algorithmName = challenge.parameters["algorithm"]
-                val algorithm =
-                    if (algorithmName == null) {
-                        DigestAlgorithm.MD5 // RFC 7616 §3.3: MD5 is the default when omitted.
-                    } else {
-                        DigestAlgorithm.fromString(algorithmName) ?: continue
-                    }
-                candidates.add(challenge to algorithm)
-            }
+            val candidates = challenges.mapNotNull(::toCandidate)
             // Pick the candidate whose algorithm appears earliest in our preference list.
             // Returns null when no candidate matches any preferred algorithm.
             for (preferred in preferredAlgorithms) {
@@ -133,6 +116,30 @@ public class DigestChallengeHandler
             return null
         }
 
+        /**
+         * Maps a single challenge to a satisfiable (challenge, algorithm) candidate, or
+         * null when it fails a validation gate. Each gate returns separately rather than
+         * collapsing into one composite predicate — keeping them distinct preserves which
+         * validation rejected a candidate when debugging Digest interop.
+         */
+        @Suppress("ReturnCount") // one early return per validation gate: scheme, realm, nonce, qop, algorithm
+        private fun toCandidate(
+            challenge: AuthenticateChallenge,
+        ): Pair<AuthenticateChallenge, DigestAlgorithm>? {
+            if (!challenge.scheme.equals("Digest", ignoreCase = true)) return null
+            if (challenge.parameters["realm"] == null) return null
+            if (challenge.parameters["nonce"] == null) return null
+            if (!qopSupportsAuth(challenge.parameters["qop"])) return null
+            val algorithmName = challenge.parameters["algorithm"]
+            val algorithm =
+                if (algorithmName == null) {
+                    DigestAlgorithm.MD5 // RFC 7616 §3.3: MD5 is the default when omitted.
+                } else {
+                    DigestAlgorithm.fromString(algorithmName) ?: return null
+                }
+            return challenge to algorithm
+        }
+
         /**
          * Builds the `Authorization: Digest ...` header value for a single challenge.
          * Follows RFC 7616 §3.4.6 — HA1, HA2, response — plus the standard parameter

sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/common/MediaType.kt

@@ -100,16 +100,16 @@ public data class MediaType private constructor(
         if (value.isNotEmpty() && value.all(::isTokenChar)) {
             return value
         }
-        val sb = StringBuilder(value.length + 2)
-        sb.append('"')
-        value.forEach { ch ->
-            if (ch == '\\' || ch == '"') {
-                sb.append('\\')
+        return buildString(value.length + 2) {
+            append('"')
+            value.forEach { ch ->
+                if (ch == '\\' || ch == '"') {
+                    append('\\')
+                }
+                append(ch)
             }
-            sb.append(ch)
+            append('"')
         }
-        sb.append('"')
-        return sb.toString()
     }
 
     /**

sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/context/DispatchContext.kt

@@ -61,7 +61,7 @@ public data class DispatchContext(
          * counter to it for the actual key.
          */
         private fun deriveCallKey(instrumentationContext: InstrumentationContext): String =
-            instrumentationContext.traceId.value + ":" + instrumentationContext.spanId.value
+            "${instrumentationContext.traceId.value}:${instrumentationContext.spanId.value}"
 
         /**
          * A dispatch context with a no-op instrumentation context; used when tracing is
@@ -81,6 +81,6 @@ public data class DispatchContext(
          * [DispatchContext]), so every link in the chain is collision-safe by default.
          */
         internal fun mintCallKey(instrumentationContext: InstrumentationContext): String =
-            deriveCallKey(instrumentationContext) + ":" + mintCounter.incrementAndGet()
+            "${deriveCallKey(instrumentationContext)}:${mintCounter.incrementAndGet()}"
     }
 }

sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/pipeline/AsyncHttpPipelineBuilder.kt

@@ -99,10 +99,7 @@ public class AsyncHttpPipelineBuilder(private val httpClient: AsyncHttpClient) {
     /** Builds an immutable [AsyncHttpPipeline]. */
     public fun build(): AsyncHttpPipeline {
         val ordered = steps.flatten()
-        val array = arrayOfNulls<AsyncHttpStep>(ordered.size)
-        for ((i, s) in ordered.withIndex()) array[i] = s
-        @Suppress("UNCHECKED_CAST")
-        return AsyncHttpPipeline(httpClient, array as Array<AsyncHttpStep>)
+        return AsyncHttpPipeline(httpClient, Array(ordered.size) { ordered[it] })
     }
 
     public companion object {

sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/pipeline/HttpPipelineBuilder.kt

@@ -110,16 +110,10 @@ public class HttpPipelineBuilder(private val httpClient: HttpClient) {
     /**
      * Builds an immutable [HttpPipeline] in stage order. [Stage.SEND] is reserved for the
      * transport and is skipped.
-     *
-     * `arrayOfNulls<HttpStep>` then fill — Kotlin's `List.toTypedArray<T>()` is erased to
-     * `Array<Any?>` at runtime which fails the `Array<HttpStep>` cast.
      */
     public fun build(): HttpPipeline {
         val ordered = steps.flatten()
-        val array = arrayOfNulls<HttpStep>(ordered.size)
-        for ((i, s) in ordered.withIndex()) array[i] = s
-        @Suppress("UNCHECKED_CAST")
-        return HttpPipeline(httpClient, array as Array<HttpStep>)
+        return HttpPipeline(httpClient, Array(ordered.size) { ordered[it] })
     }
 
     public companion object {

sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/pipeline/steps/DefaultRedirectStep.kt

@@ -258,11 +258,9 @@ public open class DefaultRedirectStep
             // underlying store may return names in mixed case (`Content-Type`), so lower-case
             // before the prefix test. Iterate a snapshot of the keys to avoid concurrent
             // modification while mutating the builder.
-            val toRemove = ArrayList<String>()
-            for (name in headers.names()) {
-                if (name.lowercase(Locale.US).startsWith("content-")) toRemove.add(name)
-            }
-            for (name in toRemove) builder.remove(name)
+            headers.names()
+                .filter { it.lowercase(Locale.US).startsWith("content-") }
+                .forEach { builder.remove(it) }
             return builder.build()
         }
 

sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/response/LoggableResponseBody.kt

@@ -122,8 +122,7 @@ public class LoggableResponseBody
          * otherwise the delegate's reported length (the true length), since the capture is just
          * a bounded prefix.
          */
-        override fun contentLength(): Long =
-            if (fullyCaptured) captured?.size ?: delegate.contentLength() else delegate.contentLength()
+        override fun contentLength(): Long = (if (fullyCaptured) captured?.size else null) ?: delegate.contentLength()
 
         /**
          * Returns a view of the captured body. Drains (up to the cap) on first call. If the drain