@@ -750,6 +750,51 @@ class HttpPipelineTest {
         assertEquals(2, copy.steps.size)
     }
 
+    @Test
+    fun `surgical insert that would duplicate a pillar throws and leaves the builder intact`() {
+        // insertAfter re-buckets by reloading the flattened step list. A second distinct RETRY-stage
+        // step would create a duplicate RETRY pillar; the reload is atomic, so the collision throws
+        // and the builder's prior steps survive — build() still yields exactly the original pillar.
+        val client = RecordingHttpClient()
+        val originalRetry = DefaultRetryStep()
+        val builder = HttpPipelineBuilder(client).append(originalRetry)
+
+        assertFailsWith<IllegalStateException> {
+            builder.insertAfter<DefaultRetryStep>(DefaultRetryStep())
+        }
+
+        val pipeline = builder.build()
+        assertEquals(1, pipeline.steps.size, "the rejected insert must not leave a duplicate pillar behind")
+        assertSame(originalRetry, pipeline.steps[0], "the original retry pillar must remain intact")
+    }
+
+    @Test
+    fun `HttpPipelineBuilder of an HttpPipeline flattens rather than nesting`() {
+        // A statically-typed HttpPipeline argument resolves to the flattening constructor: the
+        // source pipeline's steps are copied into the new builder (same count/order as from()), NOT
+        // nested as an opaque transport — which would leave the outer builder with zero steps.
+        val client = RecordingHttpClient()
+        val order = mutableListOf<String>()
+        val retryStep = TaggingStep(Stage.RETRY, "retry", order)
+        val preAuthStep = TaggingStep(Stage.PRE_AUTH, "pre-auth", order)
+        val base = HttpPipelineBuilder(client).append(retryStep).append(preAuthStep).build()
+
+        val viaConstructor = HttpPipelineBuilder(base).build()
+        val viaFrom = HttpPipelineBuilder.from(base).build()
+
+        // Same step count/order as from() — proving flatten, not nest.
+        assertEquals(viaFrom.steps.size, viaConstructor.steps.size)
+        assertEquals(2, viaConstructor.steps.size)
+        assertSame(retryStep, viaConstructor.steps[0])
+        assertSame(preAuthStep, viaConstructor.steps[1])
+        // The transport is the base's transport, not the base pipeline nested as a transport.
+        assertSame(client, viaConstructor.httpClient)
+
+        // Both steps actually run when sent (a nested pipeline would run zero outer steps).
+        viaConstructor.send(request())
+        assertEquals(listOf("retry", "pre-auth"), order)
+    }
+
     @Test
     fun `appendAll with an empty list is a no-op`() {
         val client = RecordingHttpClient()
 
@@ -87,6 +87,23 @@ class StagedStepsTest {
         assertTrue(staged.pillarAt(Stage.PRE_AUTH) == null)
     }
 
+    @Test
+    fun `reload that hits a pillar collision leaves prior state intact`() {
+        // reload commits atomically: it assembles the new state into scratch collections and only
+        // reassigns the live storage once the whole rebuild succeeds. A mid-rebuild pillar collision
+        // must therefore throw with the pre-reload state completely untouched — never half-reloaded.
+        val staged = stagedSteps()
+        staged.append(FakeStep(Stage.RETRY, "retry"))
+        staged.append(FakeStep(Stage.PRE_AUTH, "pre"))
+
+        assertFailsWith<IllegalStateException> {
+            staged.reload(listOf(FakeStep(Stage.RETRY, "retry-A"), FakeStep(Stage.RETRY, "retry-B")))
+        }
+
+        assertEquals("retry", staged.pillarAt(Stage.RETRY)?.tag, "the original pillar must survive the failed reload")
+        assertEquals(listOf("pre"), staged.nonPillarAt(Stage.PRE_AUTH).map { it.tag }, "non-pillar steps must survive")
+    }
+
     // -------- append / prepend pillar routing --------
 
     @Test
 
@@ -589,6 +589,34 @@ class AuthStepTest {
         assertEquals("refreshed-key", fake.requests[1].headers.get(HttpHeaderName.AUTHORIZATION))
     }
 
+    @Test
+    fun `401 challenge whose replacement carries a non-replayable body surfaces the real 401`() {
+        // The challenge handler returns a replacement request whose body cannot be replayed. Driving
+        // it through the chain would trip the body's consume-once guard and surface an
+        // IOException("HTTP pipeline failure") that masks the real 401. The step must instead skip
+        // the replay (like the retry/redirect steps' replayability gate) and return the original 401.
+        val step =
+            object : KeyCredentialAuthStep(KeyCredential("k")) {
+                override fun authorizeRequestOnChallenge(
+                    request: Request,
+                    response: Response,
+                ): Request =
+                    request.newBuilder()
+                        .method(Method.POST)
+                        .body(NonReplayableRequestBody())
+                        .build()
+            }
+        val fake =
+            FakeHttpClient()
+                .enqueue { status(401).header("WWW-Authenticate", "Bearer realm=\"x\"") }
+
+        val pipeline = HttpPipelineBuilder(fake).append(step).build()
+        val response = pipeline.send(getHttpsRequest())
+
+        assertEquals(401, response.status.code, "the real 401 must surface, not a masked IOException")
+        assertEquals(1, fake.callCount, "a non-replayable replacement body must skip the challenge replay")
+    }
+
     @Test
     fun `challenge retry closes the 401 response before driving the retry`() {
         val closes = AtomicInteger(0)
@@ -976,6 +1004,20 @@ class AuthStepTest {
             .url("https://api.example.com/x")
             .build()
 
+    /** Single-use request body: isReplayable() is false and writeTo streams fixed bytes. */
+    private class NonReplayableRequestBody : org.dexpace.sdk.core.http.request.RequestBody() {
+        override fun mediaType(): org.dexpace.sdk.core.http.common.MediaType? =
+            org.dexpace.sdk.core.http.common.MediaType.parse("text/plain")
+
+        override fun contentLength(): Long = 5
+
+        override fun isReplayable(): Boolean = false
+
+        override fun writeTo(sink: org.dexpace.sdk.core.io.BufferedSink) {
+            sink.write("hello".toByteArray(Charsets.UTF_8))
+        }
+    }
+
     /** ResponseBody that increments [closes] on close() and refuses source() reads. */
     private class CloseTrackingResponseBody(
         private val closes: AtomicInteger,
 
@@ -28,6 +28,7 @@ import org.dexpace.sdk.core.util.Clock
 import org.dexpace.sdk.core.util.Futures
 import java.io.IOException
 import java.io.InterruptedIOException
+import java.net.SocketTimeoutException
 import java.time.Duration
 import java.time.Instant
 import java.util.concurrent.CompletableFuture
@@ -190,6 +191,22 @@ class DefaultAsyncRetryStepTest {
         assertEquals(3, client.callCount)
     }
 
+    @Test
+    fun `SocketTimeoutException is retried as a transient failure not treated as cancellation`() {
+        // SocketTimeoutException extends InterruptedIOException, but it is a read timeout, not a
+        // cancellation — it must retry via the normal classifier, unlike the InterruptedIOException
+        // and bare-InterruptedException cases which abort after one attempt and restore the
+        // interrupt flag.
+        val client = FailNTimesClient(failures = 2) { SocketTimeoutException("read timed out") }
+        val future =
+            pipeline(client, HttpRetryOptions.fixed(maxRetries = 3, delay = Duration.ZERO))
+                .sendAsync(getRequest())
+        scheduler.runAll()
+        assertEquals(200, future.join().status.code)
+        assertEquals(3, client.callCount, "SocketTimeoutException must be retried, not surfaced as cancellation")
+        assertFalse(Thread.interrupted(), "a read timeout must not restore the interrupt flag")
+    }
+
     @Test
     fun `non-retryable exception is surfaced immediately`() {
         val client = FailNTimesClient(failures = 5) { IllegalArgumentException("nope") }
 
@@ -35,6 +35,7 @@ import java.io.ByteArrayInputStream
 import java.io.IOException
 import java.io.InputStream
 import java.io.InterruptedIOException
+import java.net.SocketTimeoutException
 import java.time.Duration
 import java.time.Instant
 import java.util.concurrent.TimeoutException
@@ -504,6 +505,32 @@ class RetryStepTest {
         assertEquals(2, attempts.get())
     }
 
+    @Test
+    fun `SocketTimeoutException is retried as a transient failure not treated as cancellation`() {
+        // SocketTimeoutException extends InterruptedIOException, but it is a read timeout — NOT a
+        // cancellation. It must be routed through the normal retry classifier (retried up to
+        // maxRetries), NOT the interrupt path that rethrows immediately and re-sets the interrupt
+        // flag (contrast `InterruptedIOException from the transport is rethrown immediately`).
+        val attempts = AtomicInteger(0)
+        val client =
+            object : HttpClient {
+                override fun execute(request: Request): Response {
+                    val n = attempts.incrementAndGet()
+                    if (n < 3) throw SocketTimeoutException("read timed out")
+                    return okResponse(request)
+                }
+            }
+        val pipeline =
+            HttpPipelineBuilder(client)
+                .append(DefaultRetryStep(HttpRetryOptions(maxRetries = 3), zeroDelayClock()))
+                .build()
+
+        val response = pipeline.send(getRequest())
+        assertEquals(200, response.status.code)
+        assertEquals(3, attempts.get(), "SocketTimeoutException must be retried, not surfaced as cancellation")
+        assertTrue(!Thread.currentThread().isInterrupted, "a read timeout must not set the interrupt flag")
+    }
+
     @Test
     fun `nested cause chain RuntimeException-of-IOException is retried`() {
         val attempts = AtomicInteger(0)
 
@@ -133,6 +133,24 @@ class ResponseExtensionsTest {
         assertEquals(bodyText, bodyString)
     }
 
+    @Test
+    fun `throwOnError buffered body is replayable and readable more than once`() {
+        // bufferErrorBody produces a replayable in-memory body: each source() call re-reads the full
+        // bytes. So the caught exception's body can be decoded via bodyAs AND then read again via
+        // string(), both seeing the complete buffered payload.
+        val bodyText = """{"error":"boom","detail":"buffered and replayable"}"""
+        val r = response(Status.INTERNAL_SERVER_ERROR, bodyText)
+        val ex = assertFailsWith<HttpException> { r.throwOnError() }
+
+        // First read: decode via bodyAs using a deserializer that consumes the whole stream.
+        val firstRead = ex.bodyAs(streamReadingDeserializer(), String::class.java)
+        assertEquals(bodyText, firstRead)
+
+        // Second read: the buffered body is replayable, so string() re-reads the full bytes.
+        val secondRead = assertNotNull(ex.body).string()
+        assertEquals(bodyText, secondRead)
+    }
+
     @Test
     fun `throwOnError exception status matches original response status`() {
         val r = response(Status.NOT_FOUND, "nope")
@@ -324,6 +342,26 @@ class ResponseExtensionsTest {
                 )
         }
 
+    /** Deserializer that reads the whole input stream and returns it as a UTF-8 String. */
+    @Suppress("UNCHECKED_CAST")
+    private fun streamReadingDeserializer(): Deserializer =
+        object : Deserializer {
+            override fun <R> deserialize(
+                input: String,
+                type: Class<R>,
+            ): R = throw UnsupportedOperationException()
+
+            override fun <R> deserialize(
+                input: ByteArray,
+                type: Class<R>,
+            ): R = throw UnsupportedOperationException()
+
+            override fun <R> deserialize(
+                inputStream: InputStream,
+                type: Class<R>,
+            ): R = inputStream.readBytes().toString(Charsets.UTF_8) as R
+        }
+
     private fun fakeSerde(deserializer: Deserializer): Serde =
         object : Serde {
             override val serializer: Serializer get() = throw UnsupportedOperationException()
 
@@ -403,6 +403,52 @@ class ResponseTest {
                 .build()
         assertFalse(resp.isSuccessful)
     }
+
+    // ---- status-class delegation (isInformational / isRedirect / isClientError / …) ---------
+
+    private fun responseWith(status: Status): Response =
+        Response.builder()
+            .request(request())
+            .protocol(Protocol.HTTP_1_1)
+            .status(status)
+            .build()
+
+    @Test
+    fun `status-class predicates delegate to Status for representative codes`() {
+        // Each predicate must mirror Status's own classification. Representative codes:
+        // 100 informational, 301 redirect, 404 client error, 503 server error, 200 success.
+        val informational = responseWith(Status.fromCode(100))
+        assertTrue(informational.isInformational)
+        assertFalse(informational.isRedirect)
+        assertFalse(informational.isClientError)
+        assertFalse(informational.isServerError)
+        assertFalse(informational.isError)
+
+        val redirect = responseWith(Status.fromCode(301))
+        assertFalse(redirect.isInformational)
+        assertTrue(redirect.isRedirect)
+        assertFalse(redirect.isClientError)
+        assertFalse(redirect.isServerError)
+        assertFalse(redirect.isError)
+
+        val clientError = responseWith(Status.fromCode(404))
+        assertFalse(clientError.isRedirect)
+        assertTrue(clientError.isClientError)
+        assertFalse(clientError.isServerError)
+        assertTrue(clientError.isError)
+
+        val serverError = responseWith(Status.fromCode(503))
+        assertFalse(serverError.isClientError)
+        assertTrue(serverError.isServerError)
+        assertTrue(serverError.isError)
+
+        val success = responseWith(Status.fromCode(200))
+        assertFalse(success.isInformational)
+        assertFalse(success.isRedirect)
+        assertFalse(success.isClientError)
+        assertFalse(success.isServerError)
+        assertFalse(success.isError)
+    }
 }
 
 class ResponseBodyTest {
@@ -522,4 +568,30 @@ class ResponseBodyTest {
         val body = ResponseBody.create(source)
         assertContentEquals(ByteArray(0), body.bytes())
     }
+
+    // ---- string() default charset derives from the media type ----------------------------
+
+    @Test
+    fun `no-arg string() decodes using the charset declared in the media type`() {
+        // 0xE9 is 'é' in ISO-8859-1 but an invalid lone byte in UTF-8. With the media type
+        // declaring charset=ISO-8859-1, the no-arg string() must latin-1-decode it to "é",
+        // not fall back to UTF-8.
+        val bytes = byteArrayOf(0xE9.toByte())
+        val mediaType = MediaType.parse("text/plain; charset=ISO-8859-1")
+        val source = Io.provider.source(bytes)
+        val body = ResponseBody.create(source, mediaType, bytes.size.toLong())
+        assertEquals("é", body.string())
+    }
+
+    @Test
+    fun `explicit charset argument overrides the media type charset`() {
+        // Bytes are the UTF-8 encoding of "é" (0xC3 0xA9). Even though the media type declares
+        // ISO-8859-1 (under which those bytes decode to "Ã©"), an explicit string(UTF_8) must use
+        // UTF-8 and yield "é" — proving the argument takes precedence over the media type charset.
+        val bytes = "é".toByteArray(Charsets.UTF_8)
+        val mediaType = MediaType.parse("text/plain; charset=ISO-8859-1")
+        val source = Io.provider.source(bytes)
+        val body = ResponseBody.create(source, mediaType, bytes.size.toLong())
+        assertEquals("é", body.string(Charsets.UTF_8))
+    }
 }
@@ -181,4 +181,37 @@ class UrlRedactorTest {
         // The trailing '&' is intentionally dropped — the output must NOT end with '&'.
         assertTrue(!out.endsWith("&"), "trailing '&' must be dropped; got: $out")
     }
+
+    // ---- redactUrlValue: absolute vs relative header-value redaction ---------------------
+
+    @Test
+    fun `redactUrlValue redacts a disallowed query credential on an absolute URL`() {
+        // An OAuth authorization code in the query of an absolute (e.g. Location) header value
+        // must be redacted just like redact(URL) does.
+        val out = UrlRedactor.redactUrlValue("https://cb/?code=SECRET")
+        assertTrue(out.contains("code="), "the query key should be kept: $out")
+        assertTrue(out.contains("***"), "the credential value should be redacted: $out")
+        assertTrue(!out.contains("SECRET"), "raw authorization code leaked: $out")
+    }
+
+    @Test
+    fun `redactUrlValue redacts an implicit-flow access token in the fragment`() {
+        // Implicit-flow tokens ride in the URL fragment as key=value tokens; they must be redacted
+        // under the same allow-list as query parameters.
+        val out = UrlRedactor.redactUrlValue("https://cb/#access_token=SECRET")
+        assertTrue(out.contains("#"), "fragment marker should be present: $out")
+        assertTrue(out.contains("access_token="), "fragment key should be kept: $out")
+        assertTrue(out.contains("***"), "fragment token value should be redacted: $out")
+        assertTrue(!out.contains("SECRET"), "raw fragment token leaked: $out")
+    }
+
+    @Test
+    fun `redactUrlValue keeps the path but drops the query for a relative value`() {
+        // A relative Location value is not a parseable absolute URL, so the whole query (which may
+        // carry an OAuth code / pre-signed signature) is dropped while the path stays visible.
+        val out = UrlRedactor.redactUrlValue("/cb?code=SECRET")
+        assertEquals("/cb?***", out)
+        assertTrue(out.startsWith("/cb"), "the path must be preserved: $out")
+        assertTrue(!out.contains("SECRET"), "raw query credential leaked from a relative value: $out")
+    }
 }