feat: add retry controls for fractional Retry-After, attempt headers, and server-driven retries

Three additive, opt-in enhancements to the retry layer:

- Parse fractional Retry-After values. RetryAfterParser now reads the
  Retry-After delta-seconds form with toDoubleOrNull, so values like
  `Retry-After: 1.5` are honoured to nanosecond resolution instead of being
  silently ignored. Integer seconds and the HTTP-date form are unchanged, and
  the negative / NaN / non-finite guards plus the 365-day ceiling still apply.
  The numeric form now shares that ceiling with the date and Unix-epoch forms,
  so a far-future delta clamps rather than flowing through unbounded. Both
  retry layers benefit, since they delegate parsing here.

- Optional per-attempt retry-count request header. RetrySettings gains an
  opt-in attemptHeaderName (default null). When set, the recovery-aware
  RetryStep stamps that header with the 1-based attempt ordinal on a per-attempt
  copy of the request, so servers and proxies can observe the retry count. The
  header is applied via Request.newBuilder, leaving the immutable template and
  any caller-supplied idempotency key untouched.

- Server-driven retry predicate. New ServerOverrideRetryPredicate implements the
  composable HttpRetryConditionPredicate seam and honours an X-Should-Retry-style
  response header (configurable name): a truthy value forces a retry, a falsy
  value suppresses one, and an absent or unrecognised value defers to a delegate
  predicate. It is not installed by default; the default retryable-status set is
  unchanged, so 409 stays out of it unless a server explicitly opts a response in.

Author: OmarAlJarrah

sdk-core/api/sdk-core.api

@@ -945,6 +945,21 @@ public abstract class org/dexpace/sdk/core/http/pipeline/steps/RetryStep : org/d
 	public final fun getStage ()Lorg/dexpace/sdk/core/http/pipeline/Stage;
 }
 
+public final class org/dexpace/sdk/core/http/pipeline/steps/ServerOverrideRetryPredicate : org/dexpace/sdk/core/http/pipeline/steps/HttpRetryConditionPredicate {
+	public static final field Companion Lorg/dexpace/sdk/core/http/pipeline/steps/ServerOverrideRetryPredicate$Companion;
+	public static final field DEFAULT_HEADER_NAME Lorg/dexpace/sdk/core/http/common/HttpHeaderName;
+	public fun <init> ()V
+	public fun <init> (Lorg/dexpace/sdk/core/http/common/HttpHeaderName;)V
+	public fun <init> (Lorg/dexpace/sdk/core/http/common/HttpHeaderName;Lorg/dexpace/sdk/core/http/pipeline/steps/HttpRetryConditionPredicate;)V
+	public synthetic fun <init> (Lorg/dexpace/sdk/core/http/common/HttpHeaderName;Lorg/dexpace/sdk/core/http/pipeline/steps/HttpRetryConditionPredicate;ILkotlin/jvm/internal/DefaultConstructorMarker;)V
+	public final fun getDelegate ()Lorg/dexpace/sdk/core/http/pipeline/steps/HttpRetryConditionPredicate;
+	public final fun getHeaderName ()Lorg/dexpace/sdk/core/http/common/HttpHeaderName;
+	public fun shouldRetry (Lorg/dexpace/sdk/core/http/pipeline/steps/HttpRetryCondition;)Z
+}
+
+public final class org/dexpace/sdk/core/http/pipeline/steps/ServerOverrideRetryPredicate$Companion {
+}
+
 public final class org/dexpace/sdk/core/http/pipeline/steps/SetDateStep : org/dexpace/sdk/core/http/pipeline/HttpStep {
 	public fun <init> ()V
 	public fun <init> (Lorg/dexpace/sdk/core/util/Clock;)V
@@ -2112,9 +2127,10 @@ public final class org/dexpace/sdk/core/pipeline/step/retry/RetrySettings {
 	public static final field DEFAULT_RETRYABLE_METHODS Ljava/util/Set;
 	public static final field DEFAULT_RETRYABLE_STATUSES Ljava/util/Set;
 	public static final field DEFAULT_TOTAL_TIMEOUT Ljava/time/Duration;
-	public synthetic fun <init> (Ljava/time/Duration;Ljava/time/Duration;DLjava/time/Duration;IDLjava/util/Set;Ljava/util/Set;Ljava/util/concurrent/ScheduledExecutorService;Lkotlin/jvm/internal/DefaultConstructorMarker;)V
+	public synthetic fun <init> (Ljava/time/Duration;Ljava/time/Duration;DLjava/time/Duration;IDLjava/util/Set;Ljava/util/Set;Ljava/util/concurrent/ScheduledExecutorService;Lorg/dexpace/sdk/core/http/common/HttpHeaderName;Lkotlin/jvm/internal/DefaultConstructorMarker;)V
 	public static final fun builder ()Lorg/dexpace/sdk/core/pipeline/step/retry/RetrySettings$RetrySettingsBuilder;
 	public static final fun defaults ()Lorg/dexpace/sdk/core/pipeline/step/retry/RetrySettings;
+	public final fun getAttemptHeaderName ()Lorg/dexpace/sdk/core/http/common/HttpHeaderName;
 	public final fun getDelayMultiplier ()D
 	public final fun getInitialDelay ()Ljava/time/Duration;
 	public final fun getJitter ()D
@@ -2135,6 +2151,7 @@ public final class org/dexpace/sdk/core/pipeline/step/retry/RetrySettings$Compan
 public final class org/dexpace/sdk/core/pipeline/step/retry/RetrySettings$RetrySettingsBuilder : org/dexpace/sdk/core/generics/Builder {
 	public fun <init> ()V
 	public fun <init> (Lorg/dexpace/sdk/core/pipeline/step/retry/RetrySettings;)V
+	public final fun attemptHeaderName (Lorg/dexpace/sdk/core/http/common/HttpHeaderName;)Lorg/dexpace/sdk/core/pipeline/step/retry/RetrySettings$RetrySettingsBuilder;
 	public synthetic fun build ()Ljava/lang/Object;
 	public fun build ()Lorg/dexpace/sdk/core/pipeline/step/retry/RetrySettings;
 	public final fun delayMultiplier (D)Lorg/dexpace/sdk/core/pipeline/step/retry/RetrySettings$RetrySettingsBuilder;

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

@@ -0,0 +1,100 @@
+/*
+ * 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.http.pipeline.steps
+
+import org.dexpace.sdk.core.http.common.HttpHeaderName
+import java.util.Locale
+
+/**
+ * Composable [HttpRetryConditionPredicate] that lets the server steer the retry decision via an
+ * explicit response header (`X-Should-Retry` by default).
+ *
+ * Some APIs annotate responses with a definitive "retry / do not retry" signal that the client
+ * cannot infer from the status code alone — for example a `409 Conflict` that is genuinely
+ * transient, or a `503` the server knows will not recover. This predicate honours that signal
+ * when, and only when, the header is present:
+ *
+ *  - A **truthy** value (`true`, `1`, `yes`, `retry`, case-insensitive) forces a retry, even for
+ *    a status the default classifier would not retry.
+ *  - A **falsy** value (`false`, `0`, `no`, `stop`, case-insensitive) suppresses a retry, even
+ *    for a status the default classifier would retry.
+ *  - When the header is **absent**, unrecognised, or there is no response (the exception path),
+ *    the decision is delegated to [delegate] — so wiring this predicate in changes behaviour
+ *    only when the server actually speaks.
+ *
+ * ## Opt-in
+ *
+ * This predicate is **not** installed by default. A caller enables it by passing an instance as
+ * [HttpRetryOptions.shouldRetryCondition] (and/or [HttpRetryOptions.shouldRetryException]). The
+ * SDK's default retryable-status set is unchanged — in particular `409 Conflict` stays out of
+ * it; this predicate is the mechanism by which a server can opt a `409` (or any other status)
+ * into a retry, rather than widening the default set for everyone.
+ *
+ * ## Composition
+ *
+ * [delegate] defaults to the SDK's standard classifier, dispatched per path: the response
+ * classifier (`408 / 429 / 5xx`, except `501` / `505`) on the response path, and the exception
+ * classifier (`IOException` / `TimeoutException` anywhere in the cause chain) on the exception
+ * path. Because the override header only appears on responses, wiring this predicate into
+ * [HttpRetryOptions.shouldRetryException] leaves the exception-path decision entirely to the
+ * delegate. Supply a different delegate to layer the server override on top of bespoke retry
+ * logic, or [HttpRetryConditionPredicate] `{ false }` to make the server header the sole
+ * authority.
+ *
+ * ## Thread-safety
+ *
+ * Immutable and stateless — safe to share across concurrent requests.
+ *
+ * @property headerName The response header consulted for the override signal. Defaults to
+ *   `X-Should-Retry`.
+ * @property delegate Fallback predicate consulted when the header is absent or unrecognised, or
+ *   on the exception path. Defaults to the standard per-path classifier.
+ */
+public class ServerOverrideRetryPredicate
+    @JvmOverloads
+    constructor(
+        public val headerName: HttpHeaderName = DEFAULT_HEADER_NAME,
+        public val delegate: HttpRetryConditionPredicate =
+            HttpRetryConditionPredicate(::defaultClassifier),
+    ) : HttpRetryConditionPredicate {
+        override fun shouldRetry(condition: HttpRetryCondition): Boolean {
+            val response = condition.response ?: return delegate.shouldRetry(condition)
+            val raw = response.headers.get(headerName) ?: return delegate.shouldRetry(condition)
+            return when (raw.trim().lowercase(Locale.US)) {
+                in TRUTHY -> true
+                in FALSY -> false
+                // An unrecognised value is not a directive — defer rather than guess.
+                else -> delegate.shouldRetry(condition)
+            }
+        }
+
+        public companion object {
+            /** Default override header (`X-Should-Retry`). */
+            @JvmField
+            public val DEFAULT_HEADER_NAME: HttpHeaderName = HttpHeaderName.fromString("X-Should-Retry")
+
+            /** Header values that force a retry. */
+            private val TRUTHY: Set<String> = setOf("true", "1", "yes", "retry")
+
+            /** Header values that suppress a retry. */
+            private val FALSY: Set<String> = setOf("false", "0", "no", "stop")
+        }
+    }
+
+/**
+ * Default [ServerOverrideRetryPredicate.delegate]: applies the SDK's standard classification for
+ * whichever path the [condition] represents — the response classifier when a response is present,
+ * the exception classifier otherwise. This keeps the override predicate's fall-through behaviour
+ * identical to the stock [HttpRetryOptions] defaults on both the response and exception paths.
+ */
+private fun defaultClassifier(condition: HttpRetryCondition): Boolean =
+    if (condition.response != null) {
+        defaultShouldRetryResponse(condition)
+    } else {
+        defaultShouldRetryException(condition)
+    }

sdk-core/src/main/kotlin/org/dexpace/sdk/core/pipeline/step/retry/RetryAfterParser.kt

@@ -28,7 +28,9 @@ import java.util.concurrent.ThreadLocalRandom
  *
  * ## Recognized header forms
  *
- *  1. `Retry-After: <seconds>` — RFC 7231 §7.1.3 numeric (delta-seconds) form.
+ *  1. `Retry-After: <seconds>` — RFC 7231 §7.1.3 numeric (delta-seconds) form. Both integer
+ *     and fractional values (e.g. `1.5`) are accepted; the fractional part is honoured to
+ *     nanosecond resolution.
  *  2. `Retry-After: <HTTP-date>` — RFC 7231 §7.1.3 absolute-date form (parsed via
  *     [DateTimeRfc1123], which tolerates an informational weekday per RFC 7231 §7.1.1.1).
  *  3. `retry-after-ms: <millis>` — millisecond delta variant.
@@ -106,6 +108,12 @@ public object RetryAfterParser {
      */
     private val MAX_DELAY: Duration = Duration.ofDays(365)
 
+    /** [MAX_DELAY] expressed in whole seconds — the clamp threshold for fractional parsing. */
+    private val MAX_DELAY_SECONDS: Double = MAX_DELAY.seconds.toDouble()
+
+    /** Nanoseconds per second — the scale factor for the fractional `Retry-After` conversion. */
+    private const val NANOS_PER_SECOND: Double = 1_000_000_000.0
+
     /**
      * Parses the next-attempt delay from [headers] relative to [now]. Returns `null` when no
      * recognized header is present or parseable.
@@ -185,14 +193,23 @@ public object RetryAfterParser {
     }
 
     /**
-     * Parses [value] as a non-negative integer count of seconds. Returns `null` on any parse
-     * failure, including negative values — the retry layer falls back to its backoff
-     * schedule rather than retrying immediately against a misbehaving server.
+     * Parses [value] as a non-negative count of seconds. Accepts both the RFC 7231 §7.1.3
+     * integer (delta-seconds) form and the fractional form (`1.5`) that real servers and
+     * proxies emit; the fractional part is honoured down to nanosecond resolution.
+     *
+     * Returns `null` on any parse failure, on a negative value, or on a non-finite value
+     * (`NaN`, `Infinity`) — the retry layer then falls back to its backoff schedule rather
+     * than retrying immediately against a misbehaving server. A finite but absurdly large
+     * value is clamped to [MAX_DELAY] before the nanosecond conversion so the resulting
+     * [Duration] can never overflow [Duration.toNanos] downstream.
      */
     private fun parseNumericSeconds(value: String): Duration? {
-        val seconds = value.toLongOrNull() ?: return null
-        if (seconds < 0L) return null
-        return Duration.ofSeconds(seconds)
+        val seconds = value.toDoubleOrNull() ?: return null
+        if (seconds.isNaN() || seconds.isInfinite() || seconds < 0.0) return null
+        // Clamp in the seconds domain before converting to nanos: a value beyond the ceiling
+        // would overflow the `* NANOS_PER_SECOND` multiply and the Long cast below.
+        if (seconds >= MAX_DELAY_SECONDS) return MAX_DELAY
+        return Duration.ofNanos((seconds * NANOS_PER_SECOND).toLong())
     }
 
     /**

sdk-core/src/main/kotlin/org/dexpace/sdk/core/pipeline/step/retry/RetrySettings.kt

@@ -8,6 +8,7 @@
 package org.dexpace.sdk.core.pipeline.step.retry
 
 import org.dexpace.sdk.core.generics.Builder
+import org.dexpace.sdk.core.http.common.HttpHeaderName
 import org.dexpace.sdk.core.http.request.Method
 import java.time.Duration
 import java.util.Collections
@@ -37,6 +38,7 @@ private val MAX_NANO_REPRESENTABLE_DELAY: Duration = Duration.ofNanos(Long.MAX_V
  *  - [retryableMethods] = `{GET, HEAD, OPTIONS, PUT, DELETE}` — safe-by-method per RFC 9110.
  *    `POST`/`PATCH`/etc. retry only when the request body is replayable.
  *  - [scheduler] = `null` — fall back to the lazy daemon scheduler created by [RetryStep].
+ *  - [attemptHeaderName] = `null` — no per-attempt header is stamped (opt-in; see the property).
  *
  * ## Thread-safety
  *
@@ -61,6 +63,12 @@ private val MAX_NANO_REPRESENTABLE_DELAY: Duration = Duration.ofNanos(Long.MAX_V
  *   RFC). Non-idempotent methods (`POST`, `PATCH`) only retry when the body is replayable.
  * @property scheduler Optional caller-provided scheduler. When `null` [RetryStep] uses a
  *   process-wide lazy daemon scheduler.
+ * @property attemptHeaderName Optional request header stamped on each attempt [RetryStep]
+ *   dispatches, carrying the 1-based attempt ordinal (`1` for the original send, `2` for the
+ *   first retry, and so on) so servers and proxies can observe the retry count. `null` (the
+ *   default) disables the header entirely. The header is set on a per-attempt copy of the
+ *   request, never on the immutable template. Any idempotency key the caller stamps stays
+ *   stable across retries — only this attempt header changes per send.
  */
 public class RetrySettings
     // The 9-arg constructor lives behind a `private` modifier — public construction goes
@@ -78,6 +86,7 @@ public class RetrySettings
         public val retryableStatuses: Set<Int>,
         public val retryableMethods: Set<Method>,
         public val scheduler: ScheduledExecutorService?,
+        public val attemptHeaderName: HttpHeaderName?,
     ) {
         /** Returns a fresh [RetrySettingsBuilder] preloaded with this instance's values. */
         public fun newBuilder(): RetrySettingsBuilder = RetrySettingsBuilder(this)
@@ -96,6 +105,7 @@ public class RetrySettings
             private var retryableStatuses: Set<Int> = DEFAULT_RETRYABLE_STATUSES
             private var retryableMethods: Set<Method> = DEFAULT_RETRYABLE_METHODS
             private var scheduler: ScheduledExecutorService? = null
+            private var attemptHeaderName: HttpHeaderName? = null
 
             /** Creates an empty builder populated with the SDK defaults. */
             public constructor()
@@ -111,6 +121,7 @@ public class RetrySettings
                 this.retryableStatuses = settings.retryableStatuses
                 this.retryableMethods = settings.retryableMethods
                 this.scheduler = settings.scheduler
+                this.attemptHeaderName = settings.attemptHeaderName
             }
 
             /** Sets [RetrySettings.totalTimeout]. Must be non-negative. */
@@ -186,6 +197,16 @@ public class RetrySettings
                     this.scheduler = scheduler
                 }
 
+            /**
+             * Sets [RetrySettings.attemptHeaderName]. When non-null, [RetryStep] stamps this
+             * header (carrying the 1-based attempt ordinal) on each attempt's request copy.
+             * `null` (the default) leaves attempts unstamped.
+             */
+            public fun attemptHeaderName(attemptHeaderName: HttpHeaderName?): RetrySettingsBuilder =
+                apply {
+                    this.attemptHeaderName = attemptHeaderName
+                }
+
             /** Builds the immutable [RetrySettings] instance. */
             override fun build(): RetrySettings =
                 RetrySettings(
@@ -198,6 +219,7 @@ public class RetrySettings
                     retryableStatuses = Collections.unmodifiableSet(LinkedHashSet(retryableStatuses)),
                     retryableMethods = Collections.unmodifiableSet(LinkedHashSet(retryableMethods)),
                     scheduler = scheduler,
+                    attemptHeaderName = attemptHeaderName,
                 )
         }
 

sdk-core/src/main/kotlin/org/dexpace/sdk/core/pipeline/step/retry/RetryStep.kt

@@ -56,6 +56,14 @@ import java.util.concurrent.TimeUnit
  * is a deliberate departure from Square's `RetryInterceptor`, which retries on status alone and
  * silently double-sends a body it cannot replay on transport timeouts.
  *
+ * ## Per-attempt header
+ *
+ * When [RetrySettings.attemptHeaderName] is configured (it is `null` by default), every send is
+ * stamped with that header carrying the 1-based attempt ordinal (`1` for the original send, `2`
+ * for the first retry, …) on a fresh per-attempt copy of the request, so servers and proxies can
+ * observe the retry count. The header is set on a copy via [Request.newBuilder]; the captured
+ * template is never mutated, and any idempotency key already on the request is left untouched.
+ *
  * ## Cancellation
  *
  * Waits between attempts use [CompletableFuture.get] backed by [scheduler] — never
@@ -142,9 +150,11 @@ public class RetryStep
          */
         @Throws(Throwable::class)
         public fun attempt(): Response {
+            // The original send is attempt ordinal 1 — stamp it when the feature is enabled so
+            // the very first request carries the same observable counter the retries will.
             val initial: ResponseOutcome =
                 try {
-                    ResponseOutcome.Success(httpClient.execute(request))
+                    ResponseOutcome.Success(httpClient.execute(stampAttempt(request, attemptOrdinal = 1)))
                 } catch (t: Throwable) {
                     ResponseOutcome.Failure(t)
                 }
@@ -172,7 +182,9 @@ public class RetryStep
                     is AttemptStep.Abort -> return ResponseOutcome.Failure(readyState.error)
                     is AttemptStep.Proceed -> {
                         state.attempt += 1
-                        val outcome = executeOnce()
+                        // state.attempt is now the 1-based ordinal of the send about to happen:
+                        // the original was 1, so the first retry dispatched here is 2.
+                        val outcome = executeOnce(state.attempt)
                         if (outcome is ResponseOutcome.Success) return outcome
                         val nextError = (outcome as ResponseOutcome.Failure).error
                         if (!isClassifiedRetryable(nextError)) return outcome
@@ -268,17 +280,39 @@ public class RetryStep
          * Executes the request once via the captured transport, converting any throwable
          * raised by the transport into a [ResponseOutcome.Failure]. Preserves the interrupt
          * flag if the transport raises an [InterruptedException] mid-send.
+         *
+         * @param attemptOrdinal The 1-based ordinal of this send, stamped onto the per-attempt
+         *  request copy when [RetrySettings.attemptHeaderName] is configured.
          */
-        private fun executeOnce(): ResponseOutcome =
+        private fun executeOnce(attemptOrdinal: Int): ResponseOutcome =
             try {
-                ResponseOutcome.Success(httpClient.execute(request))
+                ResponseOutcome.Success(httpClient.execute(stampAttempt(request, attemptOrdinal)))
             } catch (e: InterruptedException) {
                 Thread.currentThread().interrupt()
                 ResponseOutcome.Failure(e)
             } catch (t: Throwable) {
                 ResponseOutcome.Failure(t)
             }
 
+        /**
+         * Returns a per-attempt copy of [request] carrying the configured
+         * [RetrySettings.attemptHeaderName] set to [attemptOrdinal]. When no attempt header is
+         * configured the original [request] is returned unchanged — the no-op path allocates
+         * nothing, so the common (feature-off) case pays no cost. The returned copy is built via
+         * [Request.newBuilder] so the immutable template the step captured is never mutated; any
+         * idempotency key the caller stamped is preserved verbatim because only this single
+         * header is replaced.
+         */
+        private fun stampAttempt(
+            request: Request,
+            attemptOrdinal: Int,
+        ): Request {
+            val header = settings.attemptHeaderName ?: return request
+            return request.newBuilder()
+                .setHeader(header.caseSensitiveName, attemptOrdinal.toString())
+                .build()
+        }
+
         /**
          * Blocks the calling thread for [delay] without pinning a carrier thread under Loom.
          * Uses [ScheduledExecutorService.schedule] + [CompletableFuture.get] so the wait can