From 989a343bc70c60be6f6a90076a6ed2085b4c641c Mon Sep 17 00:00:00 2001
From: Alessandra Romero <romero.alessandra1@gmail.com>
Date: Fri, 22 Aug 2025 14:19:22 -0400
Subject: [PATCH 01/11] Add standard retries and unit tests

---
 .../smithy-core/src/smithy_core/aio/client.py |   6 +-
 .../src/smithy_core/interfaces/retries.py     |   6 +-
 .../smithy-core/src/smithy_core/retries.py    | 160 ++++++++++++-
 .../smithy-core/tests/unit/test_retries.py    | 213 ++++++++++++++++--
 4 files changed, 362 insertions(+), 23 deletions(-)

diff --git a/packages/smithy-core/src/smithy_core/aio/client.py b/packages/smithy-core/src/smithy_core/aio/client.py
index bf27c440c..e446f2406 100644
--- a/packages/smithy-core/src/smithy_core/aio/client.py
+++ b/packages/smithy-core/src/smithy_core/aio/client.py
@@ -330,7 +330,7 @@ async def _retry[I: SerializeableShape, O: DeserializeableShape](
             return await self._handle_attempt(call, request_context, request_future)
 
         retry_strategy = call.retry_strategy
-        retry_token = retry_strategy.acquire_initial_retry_token(
+        retry_token = await retry_strategy.acquire_initial_retry_token(
             token_scope=call.retry_scope
         )
 
@@ -349,7 +349,7 @@ async def _retry[I: SerializeableShape, O: DeserializeableShape](
 
             if isinstance(output_context.response, Exception):
                 try:
-                    retry_strategy.refresh_retry_token_for_retry(
+                    retry_token = await retry_strategy.refresh_retry_token_for_retry(
                         token_to_renew=retry_token,
                         error=output_context.response,
                     )
@@ -364,7 +364,7 @@ async def _retry[I: SerializeableShape, O: DeserializeableShape](
 
                 await seek(request_context.transport_request.body, 0)
             else:
-                retry_strategy.record_success(token=retry_token)
+                await retry_strategy.record_success(token=retry_token)
                 return output_context
 
     async def _handle_attempt[I: SerializeableShape, O: DeserializeableShape](
diff --git a/packages/smithy-core/src/smithy_core/interfaces/retries.py b/packages/smithy-core/src/smithy_core/interfaces/retries.py
index a5c9d428b..ab7bbdeed 100644
--- a/packages/smithy-core/src/smithy_core/interfaces/retries.py
+++ b/packages/smithy-core/src/smithy_core/interfaces/retries.py
@@ -61,7 +61,7 @@ class RetryStrategy(Protocol):
     max_attempts: int
     """Upper limit on total attempt count (initial attempt plus retries)."""
 
-    def acquire_initial_retry_token(
+    async def acquire_initial_retry_token(
         self, *, token_scope: str | None = None
     ) -> RetryToken:
         """Called before any retries (for the first attempt at the operation).
@@ -74,7 +74,7 @@ def acquire_initial_retry_token(
         """
         ...
 
-    def refresh_retry_token_for_retry(
+    async def refresh_retry_token_for_retry(
         self, *, token_to_renew: RetryToken, error: Exception
     ) -> RetryToken:
         """Replace an existing retry token from a failed attempt with a new token.
@@ -91,7 +91,7 @@ def refresh_retry_token_for_retry(
         """
         ...
 
-    def record_success(self, *, token: RetryToken) -> None:
+    async def record_success(self, *, token: RetryToken) -> None:
         """Return token after successful completion of an operation.
 
         Upon successful completion of the operation, a user calls this function to
diff --git a/packages/smithy-core/src/smithy_core/retries.py b/packages/smithy-core/src/smithy_core/retries.py
index 06bf6f988..c79d6b3ac 100644
--- a/packages/smithy-core/src/smithy_core/retries.py
+++ b/packages/smithy-core/src/smithy_core/retries.py
@@ -1,5 +1,6 @@
 #  Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 #  SPDX-License-Identifier: Apache-2.0
+import asyncio
 import random
 from collections.abc import Callable
 from dataclasses import dataclass
@@ -204,7 +205,7 @@ def __init__(
         self.backoff_strategy = backoff_strategy or ExponentialRetryBackoffStrategy()
         self.max_attempts = max_attempts
 
-    def acquire_initial_retry_token(
+    async def acquire_initial_retry_token(
         self, *, token_scope: str | None = None
     ) -> SimpleRetryToken:
         """Called before any retries (for the first attempt at the operation).
@@ -214,7 +215,7 @@ def acquire_initial_retry_token(
         retry_delay = self.backoff_strategy.compute_next_backoff_delay(0)
         return SimpleRetryToken(retry_count=0, retry_delay=retry_delay)
 
-    def refresh_retry_token_for_retry(
+    async def refresh_retry_token_for_retry(
         self,
         *,
         token_to_renew: retries_interface.RetryToken,
@@ -240,5 +241,158 @@ def refresh_retry_token_for_retry(
         else:
             raise RetryError(f"Error is not retryable: {error}") from error
 
-    def record_success(self, *, token: retries_interface.RetryToken) -> None:
+    async def record_success(self, *, token: retries_interface.RetryToken) -> None:
         """Not used by this retry strategy."""
+
+
+@dataclass(kw_only=True)
+class StandardRetryToken:
+    retry_count: int
+    """Retry count is the total number of attempts minus the initial attempt."""
+
+    retry_delay: float
+    """Delay in seconds to wait before the retry attempt."""
+
+    quota_consumed: int = 0
+    """The total amount of quota consumed."""
+
+    last_quota_acquired: int = 0
+    """The amount of last quota acquired."""
+
+
+class StandardRetryStrategy(retries_interface.RetryStrategy):
+    def __init__(self, *, max_attempts: int = 3):
+        """Standard retry strategy using truncated binary exponential backoff with full
+        jitter.
+
+        :param max_attempts: Upper limit on total number of attempts made, including
+            initial attempt and retries.
+        """
+        self.backoff_strategy = ExponentialRetryBackoffStrategy(
+            backoff_scale_value=1,
+            jitter_type=ExponentialBackoffJitterType.FULL,
+        )
+        self.max_attempts = max_attempts
+        self._retry_quota = StandardRetryQuota()
+
+    async def acquire_initial_retry_token(
+        self, *, token_scope: str | None = None
+    ) -> StandardRetryToken:
+        """Called before any retries (for the first attempt at the operation).
+
+        :param token_scope: This argument is ignored by this retry strategy.
+        """
+        retry_delay = self.backoff_strategy.compute_next_backoff_delay(0)
+        return StandardRetryToken(retry_count=0, retry_delay=retry_delay)
+
+    async def refresh_retry_token_for_retry(
+        self,
+        *,
+        token_to_renew: StandardRetryToken,
+        error: Exception,
+    ) -> StandardRetryToken:
+        """Replace an existing retry token from a failed attempt with a new token.
+
+        This retry strategy always returns a token until the attempt count stored in
+        the new token exceeds the ``max_attempts`` value.
+
+        :param token_to_renew: The token used for the previous failed attempt.
+        :param error: The error that triggered the need for a retry.
+        :raises RetryError: If no further retry attempts are allowed.
+        """
+        if isinstance(error, retries_interface.ErrorRetryInfo) and error.is_retry_safe:
+            retry_count = token_to_renew.retry_count + 1
+            if retry_count >= self.max_attempts:
+                raise RetryError(
+                    f"Reached maximum number of allowed attempts: {self.max_attempts}"
+                ) from error
+
+            # Acquire additional quota for this retry attempt
+            # (may raise a RetryError if none is available)
+            quota_acquired = await self._retry_quota.acquire(error=error)
+            total_quota = token_to_renew.quota_consumed + quota_acquired
+
+            if error.retry_after is not None:
+                retry_delay = error.retry_after
+            else:
+                retry_delay = self.backoff_strategy.compute_next_backoff_delay(
+                    retry_count
+                )
+
+            return StandardRetryToken(
+                retry_count=retry_count,
+                retry_delay=retry_delay,
+                quota_consumed=total_quota,
+                last_quota_acquired=quota_acquired,
+            )
+        else:
+            raise RetryError(f"Error is not retryable: {error}") from error
+
+    async def record_success(self, *, token: StandardRetryToken) -> None:
+        """Return token after successful completion of an operation.
+
+        Releases retry tokens back to the retry quota based on the previous amount
+        consumed.
+
+        :param token: The token used for the previous successful attempt.
+        """
+        await self._retry_quota.release(release_amount=token.last_quota_acquired)
+
+
+class StandardRetryQuota:
+    """Retry quota used by :py:class:`StandardRetryStrategy`."""
+
+    INITIAL_RETRY_TOKENS = 500
+    RETRY_COST = 5
+    NO_RETRY_INCREMENT = 1
+    TIMEOUT_RETRY_COST = 10
+
+    def __init__(self):
+        self._max_capacity = self.INITIAL_RETRY_TOKENS
+        self._available_capacity = self.INITIAL_RETRY_TOKENS
+        self._lock = asyncio.Lock()
+
+    async def acquire(self, *, error: Exception) -> int:
+        """Attempt to acquire a certain amount of capacity.
+
+        If there's no sufficient amount of capacity available, raise an exception.
+        Otherwise, we return the amount of capacity successfully allocated.
+        """
+        # TODO: update `is_timeout` when `is_timeout_error` is implemented
+        is_timeout = False
+        capacity_amount = self.TIMEOUT_RETRY_COST if is_timeout else self.RETRY_COST
+
+        async with self._lock:
+            if capacity_amount > self._available_capacity:
+                raise RetryError("Retry quota exceeded")
+            self._available_capacity -= capacity_amount
+            return capacity_amount
+
+    async def release(self, *, release_amount: int) -> None:
+        """Release capacity back to the retry quota.
+
+        The capacity being released will be truncated if necessary to ensure the max
+        capacity is never exceeded.
+        """
+        increment = self.NO_RETRY_INCREMENT if release_amount == 0 else release_amount
+
+        if self._available_capacity == self._max_capacity:
+            return
+
+        async with self._lock:
+            self._available_capacity = min(
+                self._available_capacity + increment, self._max_capacity
+            )
+
+
+class RetryStrategyMode(Enum):
+    """Enumeration of available retry strategies."""
+
+    SIMPLE = "simple"
+    STANDARD = "standard"
+
+
+RETRY_MODE_MAP = {
+    RetryStrategyMode.SIMPLE: SimpleRetryStrategy,
+    RetryStrategyMode.STANDARD: StandardRetryStrategy,
+}
diff --git a/packages/smithy-core/tests/unit/test_retries.py b/packages/smithy-core/tests/unit/test_retries.py
index 0b3c23be4..48b3b9286 100644
--- a/packages/smithy-core/tests/unit/test_retries.py
+++ b/packages/smithy-core/tests/unit/test_retries.py
@@ -4,7 +4,12 @@
 import pytest
 from smithy_core.exceptions import CallError, RetryError
 from smithy_core.retries import ExponentialBackoffJitterType as EBJT
-from smithy_core.retries import ExponentialRetryBackoffStrategy, SimpleRetryStrategy
+from smithy_core.retries import (
+    ExponentialRetryBackoffStrategy,
+    SimpleRetryStrategy,
+    StandardRetryQuota,
+    StandardRetryStrategy,
+)
 
 
 @pytest.mark.parametrize(
@@ -54,49 +59,229 @@ def test_exponential_backoff_strategy(
         assert delay_actual == pytest.approx(delay_expected)  # type: ignore
 
 
+@pytest.mark.asyncio
 @pytest.mark.parametrize("max_attempts", [2, 3, 10])
-def test_simple_retry_strategy(max_attempts: int) -> None:
+async def test_simple_retry_strategy(max_attempts: int) -> None:
     strategy = SimpleRetryStrategy(
         backoff_strategy=ExponentialRetryBackoffStrategy(backoff_scale_value=5),
         max_attempts=max_attempts,
     )
     error = CallError(is_retry_safe=True)
-    token = strategy.acquire_initial_retry_token()
+    token = await strategy.acquire_initial_retry_token()
     for _ in range(max_attempts - 1):
-        token = strategy.refresh_retry_token_for_retry(
+        token = await strategy.refresh_retry_token_for_retry(
             token_to_renew=token, error=error
         )
     with pytest.raises(RetryError):
-        strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
+        await strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
 
 
-def test_simple_retry_does_not_retry_unclassified() -> None:
+@pytest.mark.asyncio
+async def test_simple_retry_does_not_retry_unclassified() -> None:
     strategy = SimpleRetryStrategy(
         backoff_strategy=ExponentialRetryBackoffStrategy(backoff_scale_value=5),
         max_attempts=2,
     )
-    token = strategy.acquire_initial_retry_token()
+    token = await strategy.acquire_initial_retry_token()
     with pytest.raises(RetryError):
-        strategy.refresh_retry_token_for_retry(token_to_renew=token, error=Exception())
+        await strategy.refresh_retry_token_for_retry(
+            token_to_renew=token, error=Exception()
+        )
 
 
-def test_simple_retry_does_not_retry_when_safety_unknown() -> None:
+@pytest.mark.asyncio
+async def test_simple_retry_does_not_retry_when_safety_unknown() -> None:
     strategy = SimpleRetryStrategy(
         backoff_strategy=ExponentialRetryBackoffStrategy(backoff_scale_value=5),
         max_attempts=2,
     )
     error = CallError(is_retry_safe=None)
-    token = strategy.acquire_initial_retry_token()
+    token = await strategy.acquire_initial_retry_token()
     with pytest.raises(RetryError):
-        strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
+        await strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
 
 
-def test_simple_retry_does_not_retry_unsafe() -> None:
+@pytest.mark.asyncio
+async def test_simple_retry_does_not_retry_unsafe() -> None:
     strategy = SimpleRetryStrategy(
         backoff_strategy=ExponentialRetryBackoffStrategy(backoff_scale_value=5),
         max_attempts=2,
     )
     error = CallError(fault="client", is_retry_safe=False)
-    token = strategy.acquire_initial_retry_token()
+    token = await strategy.acquire_initial_retry_token()
+    with pytest.raises(RetryError):
+        await strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("max_attempts", [2, 3, 10])
+async def test_standard_retry_strategy(max_attempts: int) -> None:
+    strategy = StandardRetryStrategy(max_attempts=max_attempts)
+    error = CallError(is_retry_safe=True)
+    token = await strategy.acquire_initial_retry_token()
+    for _ in range(max_attempts - 1):
+        token = await strategy.refresh_retry_token_for_retry(
+            token_to_renew=token, error=error
+        )
+    with pytest.raises(RetryError):
+        await strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
+
+
+@pytest.mark.asyncio
+async def test_standard_retry_does_not_retry_unclassified() -> None:
+    strategy = StandardRetryStrategy()
+    token = await strategy.acquire_initial_retry_token()
     with pytest.raises(RetryError):
-        strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
+        await strategy.refresh_retry_token_for_retry(
+            token_to_renew=token, error=Exception()
+        )
+
+
+@pytest.mark.asyncio
+async def test_standard_retry_does_not_retry_when_safety_unknown() -> None:
+    strategy = StandardRetryStrategy()
+    error = CallError(is_retry_safe=None)
+    token = await strategy.acquire_initial_retry_token()
+    with pytest.raises(RetryError):
+        await strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
+
+
+@pytest.mark.asyncio
+async def test_standard_retry_does_not_retry_unsafe() -> None:
+    strategy = StandardRetryStrategy()
+    error = CallError(fault="client", is_retry_safe=False)
+    token = await strategy.acquire_initial_retry_token()
+    with pytest.raises(RetryError):
+        await strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
+
+
+@pytest.mark.asyncio
+async def test_standard_retry_strategy_respects_max_attempts() -> None:
+    strategy = StandardRetryStrategy()
+    error = CallError(is_retry_safe=True)
+    token = await strategy.acquire_initial_retry_token()
+    token = await strategy.refresh_retry_token_for_retry(
+        token_to_renew=token, error=error
+    )
+    token = await strategy.refresh_retry_token_for_retry(
+        token_to_renew=token, error=error
+    )
+    with pytest.raises(RetryError):
+        await strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
+
+
+@pytest.mark.asyncio
+async def test_retry_after_overrides_backoff() -> None:
+    strategy = StandardRetryStrategy()
+    error = CallError(is_retry_safe=True, retry_after=5)
+    token = await strategy.acquire_initial_retry_token()
+    token = await strategy.refresh_retry_token_for_retry(
+        token_to_renew=token, error=error
+    )
+    assert token.retry_delay == 5
+
+
+@pytest.mark.asyncio
+async def test_retry_quota_acquire_when_exhausted(monkeypatch) -> None:
+    monkeypatch.setattr(StandardRetryQuota, "INITIAL_RETRY_TOKENS", 5, raising=False)
+    monkeypatch.setattr(StandardRetryQuota, "RETRY_COST", 2, raising=False)
+
+    quota = StandardRetryQuota()
+    assert quota._available_capacity == 5
+
+    # First acquire: 5 -> 3
+    assert await quota.acquire(error=Exception()) == 2
+    assert quota._available_capacity == 3
+
+    # Second acquire: 3 -> 1
+    assert await quota.acquire(error=Exception()) == 2
+    assert quota._available_capacity == 1
+
+    # Third acquire needs 2 but only 1 remains -> should raise
+    with pytest.raises(RetryError):
+        await quota.acquire(error=Exception())
+    assert quota._available_capacity == 1
+
+
+@pytest.mark.asyncio
+async def test_retry_quota_release_zero_adds_increment(monkeypatch) -> None:
+    monkeypatch.setattr(StandardRetryQuota, "INITIAL_RETRY_TOKENS", 5, raising=False)
+    monkeypatch.setattr(StandardRetryQuota, "RETRY_COST", 2, raising=False)
+    monkeypatch.setattr(StandardRetryQuota, "NO_RETRY_INCREMENT", 1, raising=False)
+
+    quota = StandardRetryQuota()
+    assert quota._available_capacity == 5
+
+    # First acquire: 5 -> 3
+    assert await quota.acquire(error=Exception()) == 2
+    assert quota._available_capacity == 3
+
+    # release 0 should add NO_RETRY_INCREMENT: 3 -> 4
+    await quota.release(release_amount=0)
+    assert quota._available_capacity == 4
+
+    # Next acquire should still work: 4 -> 2
+    assert await quota.acquire(error=Exception()) == 2
+    assert quota._available_capacity == 2
+
+
+@pytest.mark.asyncio
+async def test_retry_quota_release_caps_at_max(monkeypatch) -> None:
+    monkeypatch.setattr(StandardRetryQuota, "INITIAL_RETRY_TOKENS", 10, raising=False)
+    monkeypatch.setattr(StandardRetryQuota, "RETRY_COST", 3, raising=False)
+
+    quota = StandardRetryQuota()
+    assert quota._available_capacity == 10
+
+    # Drain some capacity: 10 -> 7 -> 4
+    assert await quota.acquire(error=Exception()) == 3
+    assert quota._available_capacity == 7
+    assert await quota.acquire(error=Exception()) == 3
+    assert quota._available_capacity == 4
+
+    # Release more than needed: 4 + 8 = 12. Should cap at max = 10
+    await quota.release(release_amount=8)
+    assert quota._available_capacity == 10
+
+    # Another acquire should succeed from max: 10 -> 7
+    assert await quota.acquire(error=Exception()) == 3
+    assert quota._available_capacity == 7
+
+
+@pytest.mark.asyncio
+async def test_retry_quota_releases_last_acquired_amount(monkeypatch) -> None:
+    monkeypatch.setattr(StandardRetryQuota, "INITIAL_RETRY_TOKENS", 10, raising=False)
+    monkeypatch.setattr(StandardRetryQuota, "RETRY_COST", 5, raising=False)
+
+    strategy = StandardRetryStrategy()
+    err = CallError(is_retry_safe=True)
+    token = await strategy.acquire_initial_retry_token()
+
+    # Two retries: 10 -> 5 -> 0
+    token = await strategy.refresh_retry_token_for_retry(
+        token_to_renew=token, error=err
+    )
+    assert strategy._retry_quota._available_capacity == 5
+    token = await strategy.refresh_retry_token_for_retry(
+        token_to_renew=token, error=err
+    )
+    assert strategy._retry_quota._available_capacity == 0
+
+    # Success returns ONLY the last acquired amount -> 5
+    await strategy.record_success(token=token)
+    assert strategy._retry_quota._available_capacity == 5
+
+
+@pytest.mark.asyncio
+async def test_retry_quota_release_when_no_retry(monkeypatch) -> None:
+    monkeypatch.setattr(StandardRetryQuota, "INITIAL_RETRY_TOKENS", 10, raising=False)
+    quota = StandardRetryQuota()
+
+    await quota.acquire(error=Exception())
+    assert quota._available_capacity == 5
+    before = quota._available_capacity
+
+    await quota.release(release_amount=0)
+    # Should increment by NO_RETRY_INCREMENT = 1
+    assert quota._available_capacity == min(before + 1, quota._max_capacity)
+    assert quota._available_capacity == 6

From 006c5e26e9623e0ebd5d85ba18cacd5e5a39b553 Mon Sep 17 00:00:00 2001
From: Alessandra Romero <romero.alessandra1@gmail.com>
Date: Fri, 7 Nov 2025 10:45:46 -0500
Subject: [PATCH 02/11] Implement feedback and fix type errors

---
 .../smithy-core-breaking-20251106184528.json  |   4 +
 .../smithy-core/src/smithy_core/retries.py    |  34 ++-
 .../smithy-core/tests/unit/test_retries.py    | 194 +++++++++---------
 3 files changed, 128 insertions(+), 104 deletions(-)
 create mode 100644 packages/smithy-core/.changes/next-release/smithy-core-breaking-20251106184528.json

diff --git a/packages/smithy-core/.changes/next-release/smithy-core-breaking-20251106184528.json b/packages/smithy-core/.changes/next-release/smithy-core-breaking-20251106184528.json
new file mode 100644
index 000000000..7ad91343a
--- /dev/null
+++ b/packages/smithy-core/.changes/next-release/smithy-core-breaking-20251106184528.json
@@ -0,0 +1,4 @@
+{
+  "type": "breaking",
+  "description": "Added standard retry mode as the default retry strategy for AWS clients."
+}
\ No newline at end of file
diff --git a/packages/smithy-core/src/smithy_core/retries.py b/packages/smithy-core/src/smithy_core/retries.py
index c79d6b3ac..9cc93e728 100644
--- a/packages/smithy-core/src/smithy_core/retries.py
+++ b/packages/smithy-core/src/smithy_core/retries.py
@@ -268,8 +268,14 @@ def __init__(self, *, max_attempts: int = 3):
         :param max_attempts: Upper limit on total number of attempts made, including
             initial attempt and retries.
         """
+        if max_attempts < 1:
+            raise ValueError(
+                f"max_attempts must be a positive integer, got {max_attempts}"
+            )
+
         self.backoff_strategy = ExponentialRetryBackoffStrategy(
             backoff_scale_value=1,
+            max_backoff=20,
             jitter_type=ExponentialBackoffJitterType.FULL,
         )
         self.max_attempts = max_attempts
@@ -288,7 +294,7 @@ async def acquire_initial_retry_token(
     async def refresh_retry_token_for_retry(
         self,
         *,
-        token_to_renew: StandardRetryToken,
+        token_to_renew: retries_interface.RetryToken,
         error: Exception,
     ) -> StandardRetryToken:
         """Replace an existing retry token from a failed attempt with a new token.
@@ -300,6 +306,11 @@ async def refresh_retry_token_for_retry(
         :param error: The error that triggered the need for a retry.
         :raises RetryError: If no further retry attempts are allowed.
         """
+        if not isinstance(token_to_renew, StandardRetryToken):
+            raise TypeError(
+                f"StandardRetryStrategy requires StandardRetryToken, got {type(token_to_renew).__name__}"
+            )
+
         if isinstance(error, retries_interface.ErrorRetryInfo) and error.is_retry_safe:
             retry_count = token_to_renew.retry_count + 1
             if retry_count >= self.max_attempts:
@@ -310,7 +321,7 @@ async def refresh_retry_token_for_retry(
             # Acquire additional quota for this retry attempt
             # (may raise a RetryError if none is available)
             quota_acquired = await self._retry_quota.acquire(error=error)
-            total_quota = token_to_renew.quota_consumed + quota_acquired
+            total_quota: int = token_to_renew.quota_consumed + quota_acquired
 
             if error.retry_after is not None:
                 retry_delay = error.retry_after
@@ -328,7 +339,7 @@ async def refresh_retry_token_for_retry(
         else:
             raise RetryError(f"Error is not retryable: {error}") from error
 
-    async def record_success(self, *, token: StandardRetryToken) -> None:
+    async def record_success(self, *, token: retries_interface.RetryToken) -> None:
         """Return token after successful completion of an operation.
 
         Releases retry tokens back to the retry quota based on the previous amount
@@ -336,16 +347,20 @@ async def record_success(self, *, token: StandardRetryToken) -> None:
 
         :param token: The token used for the previous successful attempt.
         """
+        if not isinstance(token, StandardRetryToken):
+            raise TypeError(
+                f"StandardRetryStrategy requires StandardRetryToken, got {type(token).__name__}"
+            )
         await self._retry_quota.release(release_amount=token.last_quota_acquired)
 
 
 class StandardRetryQuota:
     """Retry quota used by :py:class:`StandardRetryStrategy`."""
 
-    INITIAL_RETRY_TOKENS = 500
-    RETRY_COST = 5
-    NO_RETRY_INCREMENT = 1
-    TIMEOUT_RETRY_COST = 10
+    INITIAL_RETRY_TOKENS: int = 500
+    RETRY_COST: int = 5
+    NO_RETRY_INCREMENT: int = 1
+    TIMEOUT_RETRY_COST: int = 10
 
     def __init__(self):
         self._max_capacity = self.INITIAL_RETRY_TOKENS
@@ -384,6 +399,11 @@ async def release(self, *, release_amount: int) -> None:
                 self._available_capacity + increment, self._max_capacity
             )
 
+    @property
+    def available_capacity(self) -> int:
+        """Return the amount of capacity available."""
+        return self._available_capacity
+
 
 class RetryStrategyMode(Enum):
     """Enumeration of available retry strategies."""
diff --git a/packages/smithy-core/tests/unit/test_retries.py b/packages/smithy-core/tests/unit/test_retries.py
index 48b3b9286..c63d31560 100644
--- a/packages/smithy-core/tests/unit/test_retries.py
+++ b/packages/smithy-core/tests/unit/test_retries.py
@@ -59,7 +59,6 @@ def test_exponential_backoff_strategy(
         assert delay_actual == pytest.approx(delay_expected)  # type: ignore
 
 
-@pytest.mark.asyncio
 @pytest.mark.parametrize("max_attempts", [2, 3, 10])
 async def test_simple_retry_strategy(max_attempts: int) -> None:
     strategy = SimpleRetryStrategy(
@@ -76,7 +75,6 @@ async def test_simple_retry_strategy(max_attempts: int) -> None:
         await strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
 
 
-@pytest.mark.asyncio
 async def test_simple_retry_does_not_retry_unclassified() -> None:
     strategy = SimpleRetryStrategy(
         backoff_strategy=ExponentialRetryBackoffStrategy(backoff_scale_value=5),
@@ -89,7 +87,6 @@ async def test_simple_retry_does_not_retry_unclassified() -> None:
         )
 
 
-@pytest.mark.asyncio
 async def test_simple_retry_does_not_retry_when_safety_unknown() -> None:
     strategy = SimpleRetryStrategy(
         backoff_strategy=ExponentialRetryBackoffStrategy(backoff_scale_value=5),
@@ -101,7 +98,6 @@ async def test_simple_retry_does_not_retry_when_safety_unknown() -> None:
         await strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
 
 
-@pytest.mark.asyncio
 async def test_simple_retry_does_not_retry_unsafe() -> None:
     strategy = SimpleRetryStrategy(
         backoff_strategy=ExponentialRetryBackoffStrategy(backoff_scale_value=5),
@@ -113,7 +109,6 @@ async def test_simple_retry_does_not_retry_unsafe() -> None:
         await strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
 
 
-@pytest.mark.asyncio
 @pytest.mark.parametrize("max_attempts", [2, 3, 10])
 async def test_standard_retry_strategy(max_attempts: int) -> None:
     strategy = StandardRetryStrategy(max_attempts=max_attempts)
@@ -127,7 +122,6 @@ async def test_standard_retry_strategy(max_attempts: int) -> None:
         await strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
 
 
-@pytest.mark.asyncio
 async def test_standard_retry_does_not_retry_unclassified() -> None:
     strategy = StandardRetryStrategy()
     token = await strategy.acquire_initial_retry_token()
@@ -137,7 +131,6 @@ async def test_standard_retry_does_not_retry_unclassified() -> None:
         )
 
 
-@pytest.mark.asyncio
 async def test_standard_retry_does_not_retry_when_safety_unknown() -> None:
     strategy = StandardRetryStrategy()
     error = CallError(is_retry_safe=None)
@@ -146,7 +139,6 @@ async def test_standard_retry_does_not_retry_when_safety_unknown() -> None:
         await strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
 
 
-@pytest.mark.asyncio
 async def test_standard_retry_does_not_retry_unsafe() -> None:
     strategy = StandardRetryStrategy()
     error = CallError(fault="client", is_retry_safe=False)
@@ -155,133 +147,141 @@ async def test_standard_retry_does_not_retry_unsafe() -> None:
         await strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
 
 
-@pytest.mark.asyncio
-async def test_standard_retry_strategy_respects_max_attempts() -> None:
+async def test_standard_retry_after_overrides_backoff() -> None:
     strategy = StandardRetryStrategy()
-    error = CallError(is_retry_safe=True)
+    error = CallError(is_retry_safe=True, retry_after=5.5)
     token = await strategy.acquire_initial_retry_token()
     token = await strategy.refresh_retry_token_for_retry(
         token_to_renew=token, error=error
     )
-    token = await strategy.refresh_retry_token_for_retry(
-        token_to_renew=token, error=error
-    )
-    with pytest.raises(RetryError):
-        await strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
+    assert token.retry_delay == 5.5
 
 
-@pytest.mark.asyncio
-async def test_retry_after_overrides_backoff() -> None:
+async def test_standard_retry_quota_consumed_accumulates() -> None:
     strategy = StandardRetryStrategy()
-    error = CallError(is_retry_safe=True, retry_after=5)
+    error = CallError(is_retry_safe=True)
     token = await strategy.acquire_initial_retry_token()
+
     token = await strategy.refresh_retry_token_for_retry(
         token_to_renew=token, error=error
     )
-    assert token.retry_delay == 5
+    first_consumed = token.quota_consumed
+    assert first_consumed == StandardRetryQuota.RETRY_COST
 
+    token = await strategy.refresh_retry_token_for_retry(
+        token_to_renew=token, error=error
+    )
+    assert token.quota_consumed == first_consumed + StandardRetryQuota.RETRY_COST
 
-@pytest.mark.asyncio
-async def test_retry_quota_acquire_when_exhausted(monkeypatch) -> None:
-    monkeypatch.setattr(StandardRetryQuota, "INITIAL_RETRY_TOKENS", 5, raising=False)
-    monkeypatch.setattr(StandardRetryQuota, "RETRY_COST", 2, raising=False)
 
-    quota = StandardRetryQuota()
-    assert quota._available_capacity == 5
+async def test_standard_retry_invalid_max_attempts() -> None:
+    with pytest.raises(ValueError):
+        StandardRetryStrategy(max_attempts=0)
 
-    # First acquire: 5 -> 3
-    assert await quota.acquire(error=Exception()) == 2
-    assert quota._available_capacity == 3
+    with pytest.raises(ValueError):
+        StandardRetryStrategy(max_attempts=-1)
 
-    # Second acquire: 3 -> 1
-    assert await quota.acquire(error=Exception()) == 2
-    assert quota._available_capacity == 1
 
-    # Third acquire needs 2 but only 1 remains -> should raise
-    with pytest.raises(RetryError):
-        await quota.acquire(error=Exception())
-    assert quota._available_capacity == 1
+async def test_standard_retry_record_success_without_retry() -> None:
+    strategy = StandardRetryStrategy()
+    token = await strategy.acquire_initial_retry_token()
+    initial_capacity = strategy._retry_quota.available_capacity  # pyright: ignore[reportPrivateUsage]
 
+    await strategy.record_success(token=token)
 
-@pytest.mark.asyncio
-async def test_retry_quota_release_zero_adds_increment(monkeypatch) -> None:
-    monkeypatch.setattr(StandardRetryQuota, "INITIAL_RETRY_TOKENS", 5, raising=False)
-    monkeypatch.setattr(StandardRetryQuota, "RETRY_COST", 2, raising=False)
-    monkeypatch.setattr(StandardRetryQuota, "NO_RETRY_INCREMENT", 1, raising=False)
+    # Should increment by NO_RETRY_INCREMENT
+    expected = min(
+        initial_capacity + StandardRetryQuota.NO_RETRY_INCREMENT,
+        StandardRetryQuota.INITIAL_RETRY_TOKENS,
+    )
+    assert strategy._retry_quota.available_capacity == expected  # pyright: ignore[reportPrivateUsage]
 
-    quota = StandardRetryQuota()
-    assert quota._available_capacity == 5
 
-    # First acquire: 5 -> 3
-    assert await quota.acquire(error=Exception()) == 2
-    assert quota._available_capacity == 3
+async def test_standard_retry_record_success_with_retry() -> None:
+    strategy = StandardRetryStrategy()
+    error = CallError(is_retry_safe=True)
+    token = await strategy.acquire_initial_retry_token()
 
-    # release 0 should add NO_RETRY_INCREMENT: 3 -> 4
-    await quota.release(release_amount=0)
-    assert quota._available_capacity == 4
+    token = await strategy.refresh_retry_token_for_retry(
+        token_to_renew=token, error=error
+    )
+    capacity_after_retry = strategy._retry_quota.available_capacity  # pyright: ignore[reportPrivateUsage]
 
-    # Next acquire should still work: 4 -> 2
-    assert await quota.acquire(error=Exception()) == 2
-    assert quota._available_capacity == 2
+    await strategy.record_success(token=token)
+
+    # Capacity should increase by last_quota_acquired
+    assert (
+        strategy._retry_quota.available_capacity  # pyright: ignore[reportPrivateUsage]
+        == capacity_after_retry + token.last_quota_acquired
+    )
 
 
-@pytest.mark.asyncio
-async def test_retry_quota_release_caps_at_max(monkeypatch) -> None:
+@pytest.fixture
+def retry_quota(monkeypatch: pytest.MonkeyPatch) -> StandardRetryQuota:
     monkeypatch.setattr(StandardRetryQuota, "INITIAL_RETRY_TOKENS", 10, raising=False)
     monkeypatch.setattr(StandardRetryQuota, "RETRY_COST", 3, raising=False)
+    monkeypatch.setattr(StandardRetryQuota, "NO_RETRY_INCREMENT", 1, raising=False)
+    return StandardRetryQuota()
 
-    quota = StandardRetryQuota()
-    assert quota._available_capacity == 10
 
-    # Drain some capacity: 10 -> 7 -> 4
-    assert await quota.acquire(error=Exception()) == 3
-    assert quota._available_capacity == 7
-    assert await quota.acquire(error=Exception()) == 3
-    assert quota._available_capacity == 4
+async def test_retry_quota_initial_state(
+    retry_quota: StandardRetryQuota,
+) -> None:
+    assert retry_quota.available_capacity == 10
+    assert retry_quota._max_capacity == 10  # pyright: ignore[reportPrivateUsage]
 
-    # Release more than needed: 4 + 8 = 12. Should cap at max = 10
-    await quota.release(release_amount=8)
-    assert quota._available_capacity == 10
 
-    # Another acquire should succeed from max: 10 -> 7
-    assert await quota.acquire(error=Exception()) == 3
-    assert quota._available_capacity == 7
+async def test_retry_quota_acquire_success(
+    retry_quota: StandardRetryQuota,
+) -> None:
+    acquired = await retry_quota.acquire(error=Exception())
 
+    assert acquired == 3
+    assert retry_quota.available_capacity == 7
 
-@pytest.mark.asyncio
-async def test_retry_quota_releases_last_acquired_amount(monkeypatch) -> None:
-    monkeypatch.setattr(StandardRetryQuota, "INITIAL_RETRY_TOKENS", 10, raising=False)
-    monkeypatch.setattr(StandardRetryQuota, "RETRY_COST", 5, raising=False)
 
-    strategy = StandardRetryStrategy()
-    err = CallError(is_retry_safe=True)
-    token = await strategy.acquire_initial_retry_token()
+async def test_retry_quota_acquire_when_exhausted(
+    retry_quota: StandardRetryQuota,
+) -> None:
+    # Drain capacity: 10 -> 7 -> 4 -> 1
+    await retry_quota.acquire(error=Exception())
+    await retry_quota.acquire(error=Exception())
+    await retry_quota.acquire(error=Exception())
+    assert retry_quota.available_capacity == 1
 
-    # Two retries: 10 -> 5 -> 0
-    token = await strategy.refresh_retry_token_for_retry(
-        token_to_renew=token, error=err
-    )
-    assert strategy._retry_quota._available_capacity == 5
-    token = await strategy.refresh_retry_token_for_retry(
-        token_to_renew=token, error=err
-    )
-    assert strategy._retry_quota._available_capacity == 0
+    # Next acquire needs 3 but only 1 remains
+    with pytest.raises(RetryError, match="Retry quota exceeded"):
+        await retry_quota.acquire(error=Exception())
 
-    # Success returns ONLY the last acquired amount -> 5
-    await strategy.record_success(token=token)
-    assert strategy._retry_quota._available_capacity == 5
 
+async def test_retry_quota_release_restores_capacity(
+    retry_quota: StandardRetryQuota,
+) -> None:
+    acquired = await retry_quota.acquire(error=Exception())
+    assert retry_quota.available_capacity == 7
 
-@pytest.mark.asyncio
-async def test_retry_quota_release_when_no_retry(monkeypatch) -> None:
-    monkeypatch.setattr(StandardRetryQuota, "INITIAL_RETRY_TOKENS", 10, raising=False)
-    quota = StandardRetryQuota()
+    await retry_quota.release(release_amount=acquired)
+    assert retry_quota.available_capacity == 10
 
-    await quota.acquire(error=Exception())
-    assert quota._available_capacity == 5
-    before = quota._available_capacity
 
-    await quota.release(release_amount=0)
-    # Should increment by NO_RETRY_INCREMENT = 1
-    assert quota._available_capacity == min(before + 1, quota._max_capacity)
-    assert quota._available_capacity == 6
+async def test_retry_quota_release_zero_adds_increment(
+    retry_quota: StandardRetryQuota,
+) -> None:
+    await retry_quota.acquire(error=Exception())
+    assert retry_quota.available_capacity == 7
+
+    await retry_quota.release(release_amount=0)
+    assert retry_quota.available_capacity == 8
+
+
+async def test_retry_quota_release_caps_at_max(
+    retry_quota: StandardRetryQuota,
+) -> None:
+    # Drain some capacity
+    await retry_quota.acquire(error=Exception())
+    await retry_quota.acquire(error=Exception())
+    assert retry_quota.available_capacity == 4
+
+    # Release more than drained. Should cap at max
+    await retry_quota.release(release_amount=20)
+    assert retry_quota.available_capacity == 10

From 0e5e2eb803be4507094c0b05ee607c8be002984c Mon Sep 17 00:00:00 2001
From: Alessandra Romero <romero.alessandra1@gmail.com>
Date: Wed, 12 Nov 2025 11:19:12 -0500
Subject: [PATCH 03/11] Update changelog entry

---
 .../next-release/smithy-core-breaking-20251106184528.json     | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/packages/smithy-core/.changes/next-release/smithy-core-breaking-20251106184528.json b/packages/smithy-core/.changes/next-release/smithy-core-breaking-20251106184528.json
index 7ad91343a..6fa68f5d8 100644
--- a/packages/smithy-core/.changes/next-release/smithy-core-breaking-20251106184528.json
+++ b/packages/smithy-core/.changes/next-release/smithy-core-breaking-20251106184528.json
@@ -1,4 +1,4 @@
 {
-  "type": "breaking",
-  "description": "Added standard retry mode as the default retry strategy for AWS clients."
+  "type": "feature",
+  "description": "Added support for `standard` retry mode."
 }
\ No newline at end of file

From 1cc663199477bf0de26601ff3bfafa640ca879fa Mon Sep 17 00:00:00 2001
From: Alessandra Romero <24320222+alexgromero@users.noreply.github.com>
Date: Wed, 12 Nov 2025 16:33:35 -0500
Subject: [PATCH 04/11] Update packages/smithy-core/src/smithy_core/retries.py

Co-authored-by: Nate Prewitt <nate.prewitt@gmail.com>
---
 packages/smithy-core/src/smithy_core/retries.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/packages/smithy-core/src/smithy_core/retries.py b/packages/smithy-core/src/smithy_core/retries.py
index 9cc93e728..751560ba5 100644
--- a/packages/smithy-core/src/smithy_core/retries.py
+++ b/packages/smithy-core/src/smithy_core/retries.py
@@ -370,7 +370,7 @@ def __init__(self):
     async def acquire(self, *, error: Exception) -> int:
         """Attempt to acquire a certain amount of capacity.
 
-        If there's no sufficient amount of capacity available, raise an exception.
+        If there's insufficient capacity available, raise an exception.
         Otherwise, we return the amount of capacity successfully allocated.
         """
         # TODO: update `is_timeout` when `is_timeout_error` is implemented

From fe4b444502e9ae301766b5bfc589eb32f31164c0 Mon Sep 17 00:00:00 2001
From: Alessandra Romero <romero.alessandra1@gmail.com>
Date: Thu, 13 Nov 2025 11:05:29 -0500
Subject: [PATCH 05/11] Fix docstrings

---
 .../smithy-core/src/smithy_core/interfaces/retries.py    | 2 +-
 packages/smithy-core/src/smithy_core/retries.py          | 9 +++------
 2 files changed, 4 insertions(+), 7 deletions(-)

diff --git a/packages/smithy-core/src/smithy_core/interfaces/retries.py b/packages/smithy-core/src/smithy_core/interfaces/retries.py
index ab7bbdeed..55da50081 100644
--- a/packages/smithy-core/src/smithy_core/interfaces/retries.py
+++ b/packages/smithy-core/src/smithy_core/interfaces/retries.py
@@ -64,7 +64,7 @@ class RetryStrategy(Protocol):
     async def acquire_initial_retry_token(
         self, *, token_scope: str | None = None
     ) -> RetryToken:
-        """Called before any retries (for the first attempt at the operation).
+        """Create a base retry token for the start of a request.
 
         :param token_scope: An arbitrary string accepted by the retry strategy to
             separate tokens into scopes.
diff --git a/packages/smithy-core/src/smithy_core/retries.py b/packages/smithy-core/src/smithy_core/retries.py
index 751560ba5..40731e3b7 100644
--- a/packages/smithy-core/src/smithy_core/retries.py
+++ b/packages/smithy-core/src/smithy_core/retries.py
@@ -208,7 +208,7 @@ def __init__(
     async def acquire_initial_retry_token(
         self, *, token_scope: str | None = None
     ) -> SimpleRetryToken:
-        """Called before any retries (for the first attempt at the operation).
+        """Create a base retry token for the start of a request.
 
         :param token_scope: This argument is ignored by this retry strategy.
         """
@@ -284,7 +284,7 @@ def __init__(self, *, max_attempts: int = 3):
     async def acquire_initial_retry_token(
         self, *, token_scope: str | None = None
     ) -> StandardRetryToken:
-        """Called before any retries (for the first attempt at the operation).
+        """Create a base retry token for the start of a request.
 
         :param token_scope: This argument is ignored by this retry strategy.
         """
@@ -340,10 +340,7 @@ async def refresh_retry_token_for_retry(
             raise RetryError(f"Error is not retryable: {error}") from error
 
     async def record_success(self, *, token: retries_interface.RetryToken) -> None:
-        """Return token after successful completion of an operation.
-
-        Releases retry tokens back to the retry quota based on the previous amount
-        consumed.
+        """Release retry quota back based on the amount consumed by the last retry.
 
         :param token: The token used for the previous successful attempt.
         """

From a99562e7c1e010a0dd284ff1396bc3553e346b01 Mon Sep 17 00:00:00 2001
From: Alessandra Romero <romero.alessandra1@gmail.com>
Date: Thu, 13 Nov 2025 11:08:44 -0500
Subject: [PATCH 06/11] Add backoff_strategy param for configurable
 RetryBackoffStrategy

---
 packages/smithy-core/src/smithy_core/retries.py | 16 ++++++++++++----
 1 file changed, 12 insertions(+), 4 deletions(-)

diff --git a/packages/smithy-core/src/smithy_core/retries.py b/packages/smithy-core/src/smithy_core/retries.py
index 40731e3b7..1c5b5657b 100644
--- a/packages/smithy-core/src/smithy_core/retries.py
+++ b/packages/smithy-core/src/smithy_core/retries.py
@@ -261,19 +261,27 @@ class StandardRetryToken:
 
 
 class StandardRetryStrategy(retries_interface.RetryStrategy):
-    def __init__(self, *, max_attempts: int = 3):
+    def __init__(
+        self,
+        *,
+        backoff_strategy: retries_interface.RetryBackoffStrategy | None = None,
+        max_attempts: int = 3,
+    ):
         """Standard retry strategy using truncated binary exponential backoff with full
         jitter.
 
+        :param backoff_strategy: The backoff strategy used by returned tokens to compute
+        the retry delay. Defaults to :py:class:`ExponentialRetryBackoffStrategy`.
+
         :param max_attempts: Upper limit on total number of attempts made, including
             initial attempt and retries.
         """
-        if max_attempts < 1:
+        if max_attempts < 0:
             raise ValueError(
-                f"max_attempts must be a positive integer, got {max_attempts}"
+                f"max_attempts must be a non-negative integer, got {max_attempts}"
             )
 
-        self.backoff_strategy = ExponentialRetryBackoffStrategy(
+        self.backoff_strategy = backoff_strategy or ExponentialRetryBackoffStrategy(
             backoff_scale_value=1,
             max_backoff=20,
             jitter_type=ExponentialBackoffJitterType.FULL,

From 9199b8ef55b509968f4527cc838bf717ec2c4d7e Mon Sep 17 00:00:00 2001
From: Alessandra Romero <romero.alessandra1@gmail.com>
Date: Thu, 13 Nov 2025 11:09:48 -0500
Subject: [PATCH 07/11] Remove RetryStrategyMode enum

---
 packages/smithy-core/src/smithy_core/retries.py | 13 -------------
 1 file changed, 13 deletions(-)

diff --git a/packages/smithy-core/src/smithy_core/retries.py b/packages/smithy-core/src/smithy_core/retries.py
index 1c5b5657b..fff637b71 100644
--- a/packages/smithy-core/src/smithy_core/retries.py
+++ b/packages/smithy-core/src/smithy_core/retries.py
@@ -408,16 +408,3 @@ async def release(self, *, release_amount: int) -> None:
     def available_capacity(self) -> int:
         """Return the amount of capacity available."""
         return self._available_capacity
-
-
-class RetryStrategyMode(Enum):
-    """Enumeration of available retry strategies."""
-
-    SIMPLE = "simple"
-    STANDARD = "standard"
-
-
-RETRY_MODE_MAP = {
-    RetryStrategyMode.SIMPLE: SimpleRetryStrategy,
-    RetryStrategyMode.STANDARD: StandardRetryStrategy,
-}

From 3dae8309c08b54c45cb9476cd104436505037a08 Mon Sep 17 00:00:00 2001
From: Alessandra Romero <romero.alessandra1@gmail.com>
Date: Thu, 13 Nov 2025 11:53:13 -0500
Subject: [PATCH 08/11] Make retry strategy methods synchronous with
 threading.Lock and update tests

---
 .../smithy-core/src/smithy_core/aio/client.py |   6 +-
 .../src/smithy_core/interfaces/retries.py     |   6 +-
 .../smithy-core/src/smithy_core/retries.py    |  28 ++--
 .../smithy-core/tests/unit/test_retries.py    | 133 ++++++++----------
 4 files changed, 79 insertions(+), 94 deletions(-)

diff --git a/packages/smithy-core/src/smithy_core/aio/client.py b/packages/smithy-core/src/smithy_core/aio/client.py
index e446f2406..01357cd65 100644
--- a/packages/smithy-core/src/smithy_core/aio/client.py
+++ b/packages/smithy-core/src/smithy_core/aio/client.py
@@ -330,7 +330,7 @@ async def _retry[I: SerializeableShape, O: DeserializeableShape](
             return await self._handle_attempt(call, request_context, request_future)
 
         retry_strategy = call.retry_strategy
-        retry_token = await retry_strategy.acquire_initial_retry_token(
+        retry_token = retry_strategy.acquire_initial_retry_token(
             token_scope=call.retry_scope
         )
 
@@ -349,7 +349,7 @@ async def _retry[I: SerializeableShape, O: DeserializeableShape](
 
             if isinstance(output_context.response, Exception):
                 try:
-                    retry_token = await retry_strategy.refresh_retry_token_for_retry(
+                    retry_token = retry_strategy.refresh_retry_token_for_retry(
                         token_to_renew=retry_token,
                         error=output_context.response,
                     )
@@ -364,7 +364,7 @@ async def _retry[I: SerializeableShape, O: DeserializeableShape](
 
                 await seek(request_context.transport_request.body, 0)
             else:
-                await retry_strategy.record_success(token=retry_token)
+                retry_strategy.record_success(token=retry_token)
                 return output_context
 
     async def _handle_attempt[I: SerializeableShape, O: DeserializeableShape](
diff --git a/packages/smithy-core/src/smithy_core/interfaces/retries.py b/packages/smithy-core/src/smithy_core/interfaces/retries.py
index 55da50081..f100dca32 100644
--- a/packages/smithy-core/src/smithy_core/interfaces/retries.py
+++ b/packages/smithy-core/src/smithy_core/interfaces/retries.py
@@ -61,7 +61,7 @@ class RetryStrategy(Protocol):
     max_attempts: int
     """Upper limit on total attempt count (initial attempt plus retries)."""
 
-    async def acquire_initial_retry_token(
+    def acquire_initial_retry_token(
         self, *, token_scope: str | None = None
     ) -> RetryToken:
         """Create a base retry token for the start of a request.
@@ -74,7 +74,7 @@ async def acquire_initial_retry_token(
         """
         ...
 
-    async def refresh_retry_token_for_retry(
+    def refresh_retry_token_for_retry(
         self, *, token_to_renew: RetryToken, error: Exception
     ) -> RetryToken:
         """Replace an existing retry token from a failed attempt with a new token.
@@ -91,7 +91,7 @@ async def refresh_retry_token_for_retry(
         """
         ...
 
-    async def record_success(self, *, token: RetryToken) -> None:
+    def record_success(self, *, token: RetryToken) -> None:
         """Return token after successful completion of an operation.
 
         Upon successful completion of the operation, a user calls this function to
diff --git a/packages/smithy-core/src/smithy_core/retries.py b/packages/smithy-core/src/smithy_core/retries.py
index fff637b71..ccc2bbbbb 100644
--- a/packages/smithy-core/src/smithy_core/retries.py
+++ b/packages/smithy-core/src/smithy_core/retries.py
@@ -1,7 +1,7 @@
 #  Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 #  SPDX-License-Identifier: Apache-2.0
-import asyncio
 import random
+import threading
 from collections.abc import Callable
 from dataclasses import dataclass
 from enum import Enum
@@ -205,7 +205,7 @@ def __init__(
         self.backoff_strategy = backoff_strategy or ExponentialRetryBackoffStrategy()
         self.max_attempts = max_attempts
 
-    async def acquire_initial_retry_token(
+    def acquire_initial_retry_token(
         self, *, token_scope: str | None = None
     ) -> SimpleRetryToken:
         """Create a base retry token for the start of a request.
@@ -215,7 +215,7 @@ async def acquire_initial_retry_token(
         retry_delay = self.backoff_strategy.compute_next_backoff_delay(0)
         return SimpleRetryToken(retry_count=0, retry_delay=retry_delay)
 
-    async def refresh_retry_token_for_retry(
+    def refresh_retry_token_for_retry(
         self,
         *,
         token_to_renew: retries_interface.RetryToken,
@@ -241,7 +241,7 @@ async def refresh_retry_token_for_retry(
         else:
             raise RetryError(f"Error is not retryable: {error}") from error
 
-    async def record_success(self, *, token: retries_interface.RetryToken) -> None:
+    def record_success(self, *, token: retries_interface.RetryToken) -> None:
         """Not used by this retry strategy."""
 
 
@@ -289,7 +289,7 @@ def __init__(
         self.max_attempts = max_attempts
         self._retry_quota = StandardRetryQuota()
 
-    async def acquire_initial_retry_token(
+    def acquire_initial_retry_token(
         self, *, token_scope: str | None = None
     ) -> StandardRetryToken:
         """Create a base retry token for the start of a request.
@@ -299,7 +299,7 @@ async def acquire_initial_retry_token(
         retry_delay = self.backoff_strategy.compute_next_backoff_delay(0)
         return StandardRetryToken(retry_count=0, retry_delay=retry_delay)
 
-    async def refresh_retry_token_for_retry(
+    def refresh_retry_token_for_retry(
         self,
         *,
         token_to_renew: retries_interface.RetryToken,
@@ -328,7 +328,7 @@ async def refresh_retry_token_for_retry(
 
             # Acquire additional quota for this retry attempt
             # (may raise a RetryError if none is available)
-            quota_acquired = await self._retry_quota.acquire(error=error)
+            quota_acquired = self._retry_quota.acquire(error=error)
             total_quota: int = token_to_renew.quota_consumed + quota_acquired
 
             if error.retry_after is not None:
@@ -347,7 +347,7 @@ async def refresh_retry_token_for_retry(
         else:
             raise RetryError(f"Error is not retryable: {error}") from error
 
-    async def record_success(self, *, token: retries_interface.RetryToken) -> None:
+    def record_success(self, *, token: retries_interface.RetryToken) -> None:
         """Release retry quota back based on the amount consumed by the last retry.
 
         :param token: The token used for the previous successful attempt.
@@ -356,7 +356,7 @@ async def record_success(self, *, token: retries_interface.RetryToken) -> None:
             raise TypeError(
                 f"StandardRetryStrategy requires StandardRetryToken, got {type(token).__name__}"
             )
-        await self._retry_quota.release(release_amount=token.last_quota_acquired)
+        self._retry_quota.release(release_amount=token.last_quota_acquired)
 
 
 class StandardRetryQuota:
@@ -370,9 +370,9 @@ class StandardRetryQuota:
     def __init__(self):
         self._max_capacity = self.INITIAL_RETRY_TOKENS
         self._available_capacity = self.INITIAL_RETRY_TOKENS
-        self._lock = asyncio.Lock()
+        self._lock = threading.Lock()
 
-    async def acquire(self, *, error: Exception) -> int:
+    def acquire(self, *, error: Exception) -> int:
         """Attempt to acquire a certain amount of capacity.
 
         If there's insufficient capacity available, raise an exception.
@@ -382,13 +382,13 @@ async def acquire(self, *, error: Exception) -> int:
         is_timeout = False
         capacity_amount = self.TIMEOUT_RETRY_COST if is_timeout else self.RETRY_COST
 
-        async with self._lock:
+        with self._lock:
             if capacity_amount > self._available_capacity:
                 raise RetryError("Retry quota exceeded")
             self._available_capacity -= capacity_amount
             return capacity_amount
 
-    async def release(self, *, release_amount: int) -> None:
+    def release(self, *, release_amount: int) -> None:
         """Release capacity back to the retry quota.
 
         The capacity being released will be truncated if necessary to ensure the max
@@ -399,7 +399,7 @@ async def release(self, *, release_amount: int) -> None:
         if self._available_capacity == self._max_capacity:
             return
 
-        async with self._lock:
+        with self._lock:
             self._available_capacity = min(
                 self._available_capacity + increment, self._max_capacity
             )
diff --git a/packages/smithy-core/tests/unit/test_retries.py b/packages/smithy-core/tests/unit/test_retries.py
index c63d31560..ece1a394f 100644
--- a/packages/smithy-core/tests/unit/test_retries.py
+++ b/packages/smithy-core/tests/unit/test_retries.py
@@ -60,134 +60,121 @@ def test_exponential_backoff_strategy(
 
 
 @pytest.mark.parametrize("max_attempts", [2, 3, 10])
-async def test_simple_retry_strategy(max_attempts: int) -> None:
+def test_simple_retry_strategy(max_attempts: int) -> None:
     strategy = SimpleRetryStrategy(
         backoff_strategy=ExponentialRetryBackoffStrategy(backoff_scale_value=5),
         max_attempts=max_attempts,
     )
     error = CallError(is_retry_safe=True)
-    token = await strategy.acquire_initial_retry_token()
+    token = strategy.acquire_initial_retry_token()
     for _ in range(max_attempts - 1):
-        token = await strategy.refresh_retry_token_for_retry(
+        token = strategy.refresh_retry_token_for_retry(
             token_to_renew=token, error=error
         )
     with pytest.raises(RetryError):
-        await strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
+        strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
 
 
-async def test_simple_retry_does_not_retry_unclassified() -> None:
+def test_simple_retry_does_not_retry_unclassified() -> None:
     strategy = SimpleRetryStrategy(
         backoff_strategy=ExponentialRetryBackoffStrategy(backoff_scale_value=5),
         max_attempts=2,
     )
-    token = await strategy.acquire_initial_retry_token()
+    token = strategy.acquire_initial_retry_token()
     with pytest.raises(RetryError):
-        await strategy.refresh_retry_token_for_retry(
-            token_to_renew=token, error=Exception()
-        )
+        strategy.refresh_retry_token_for_retry(token_to_renew=token, error=Exception())
 
 
-async def test_simple_retry_does_not_retry_when_safety_unknown() -> None:
+def test_simple_retry_does_not_retry_when_safety_unknown() -> None:
     strategy = SimpleRetryStrategy(
         backoff_strategy=ExponentialRetryBackoffStrategy(backoff_scale_value=5),
         max_attempts=2,
     )
     error = CallError(is_retry_safe=None)
-    token = await strategy.acquire_initial_retry_token()
+    token = strategy.acquire_initial_retry_token()
     with pytest.raises(RetryError):
-        await strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
+        strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
 
 
-async def test_simple_retry_does_not_retry_unsafe() -> None:
+def test_simple_retry_does_not_retry_unsafe() -> None:
     strategy = SimpleRetryStrategy(
         backoff_strategy=ExponentialRetryBackoffStrategy(backoff_scale_value=5),
         max_attempts=2,
     )
     error = CallError(fault="client", is_retry_safe=False)
-    token = await strategy.acquire_initial_retry_token()
+    token = strategy.acquire_initial_retry_token()
     with pytest.raises(RetryError):
-        await strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
+        strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
 
 
 @pytest.mark.parametrize("max_attempts", [2, 3, 10])
-async def test_standard_retry_strategy(max_attempts: int) -> None:
+def test_standard_retry_strategy(max_attempts: int) -> None:
     strategy = StandardRetryStrategy(max_attempts=max_attempts)
     error = CallError(is_retry_safe=True)
-    token = await strategy.acquire_initial_retry_token()
+    token = strategy.acquire_initial_retry_token()
     for _ in range(max_attempts - 1):
-        token = await strategy.refresh_retry_token_for_retry(
+        token = strategy.refresh_retry_token_for_retry(
             token_to_renew=token, error=error
         )
     with pytest.raises(RetryError):
-        await strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
+        strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
 
 
-async def test_standard_retry_does_not_retry_unclassified() -> None:
+def test_standard_retry_does_not_retry_unclassified() -> None:
     strategy = StandardRetryStrategy()
-    token = await strategy.acquire_initial_retry_token()
+    token = strategy.acquire_initial_retry_token()
     with pytest.raises(RetryError):
-        await strategy.refresh_retry_token_for_retry(
-            token_to_renew=token, error=Exception()
-        )
+        strategy.refresh_retry_token_for_retry(token_to_renew=token, error=Exception())
 
 
-async def test_standard_retry_does_not_retry_when_safety_unknown() -> None:
+def test_standard_retry_does_not_retry_when_safety_unknown() -> None:
     strategy = StandardRetryStrategy()
     error = CallError(is_retry_safe=None)
-    token = await strategy.acquire_initial_retry_token()
+    token = strategy.acquire_initial_retry_token()
     with pytest.raises(RetryError):
-        await strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
+        strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
 
 
-async def test_standard_retry_does_not_retry_unsafe() -> None:
+def test_standard_retry_does_not_retry_unsafe() -> None:
     strategy = StandardRetryStrategy()
     error = CallError(fault="client", is_retry_safe=False)
-    token = await strategy.acquire_initial_retry_token()
+    token = strategy.acquire_initial_retry_token()
     with pytest.raises(RetryError):
-        await strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
+        strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
 
 
-async def test_standard_retry_after_overrides_backoff() -> None:
+def test_standard_retry_after_overrides_backoff() -> None:
     strategy = StandardRetryStrategy()
     error = CallError(is_retry_safe=True, retry_after=5.5)
-    token = await strategy.acquire_initial_retry_token()
-    token = await strategy.refresh_retry_token_for_retry(
-        token_to_renew=token, error=error
-    )
+    token = strategy.acquire_initial_retry_token()
+    token = strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
     assert token.retry_delay == 5.5
 
 
-async def test_standard_retry_quota_consumed_accumulates() -> None:
+def test_standard_retry_quota_consumed_accumulates() -> None:
     strategy = StandardRetryStrategy()
     error = CallError(is_retry_safe=True)
-    token = await strategy.acquire_initial_retry_token()
+    token = strategy.acquire_initial_retry_token()
 
-    token = await strategy.refresh_retry_token_for_retry(
-        token_to_renew=token, error=error
-    )
+    token = strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
     first_consumed = token.quota_consumed
     assert first_consumed == StandardRetryQuota.RETRY_COST
 
-    token = await strategy.refresh_retry_token_for_retry(
-        token_to_renew=token, error=error
-    )
+    token = strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
     assert token.quota_consumed == first_consumed + StandardRetryQuota.RETRY_COST
 
 
-async def test_standard_retry_invalid_max_attempts() -> None:
-    with pytest.raises(ValueError):
-        StandardRetryStrategy(max_attempts=0)
-
+def test_standard_retry_invalid_max_attempts() -> None:
     with pytest.raises(ValueError):
         StandardRetryStrategy(max_attempts=-1)
 
 
-async def test_standard_retry_record_success_without_retry() -> None:
+def test_standard_retry_record_success_without_retry() -> None:
     strategy = StandardRetryStrategy()
-    token = await strategy.acquire_initial_retry_token()
+    token = strategy.acquire_initial_retry_token()
     initial_capacity = strategy._retry_quota.available_capacity  # pyright: ignore[reportPrivateUsage]
 
-    await strategy.record_success(token=token)
+    strategy.record_success(token=token)
 
     # Should increment by NO_RETRY_INCREMENT
     expected = min(
@@ -197,17 +184,15 @@ async def test_standard_retry_record_success_without_retry() -> None:
     assert strategy._retry_quota.available_capacity == expected  # pyright: ignore[reportPrivateUsage]
 
 
-async def test_standard_retry_record_success_with_retry() -> None:
+def test_standard_retry_record_success_with_retry() -> None:
     strategy = StandardRetryStrategy()
     error = CallError(is_retry_safe=True)
-    token = await strategy.acquire_initial_retry_token()
+    token = strategy.acquire_initial_retry_token()
 
-    token = await strategy.refresh_retry_token_for_retry(
-        token_to_renew=token, error=error
-    )
+    token = strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
     capacity_after_retry = strategy._retry_quota.available_capacity  # pyright: ignore[reportPrivateUsage]
 
-    await strategy.record_success(token=token)
+    strategy.record_success(token=token)
 
     # Capacity should increase by last_quota_acquired
     assert (
@@ -224,64 +209,64 @@ def retry_quota(monkeypatch: pytest.MonkeyPatch) -> StandardRetryQuota:
     return StandardRetryQuota()
 
 
-async def test_retry_quota_initial_state(
+def test_retry_quota_initial_state(
     retry_quota: StandardRetryQuota,
 ) -> None:
     assert retry_quota.available_capacity == 10
     assert retry_quota._max_capacity == 10  # pyright: ignore[reportPrivateUsage]
 
 
-async def test_retry_quota_acquire_success(
+def test_retry_quota_acquire_success(
     retry_quota: StandardRetryQuota,
 ) -> None:
-    acquired = await retry_quota.acquire(error=Exception())
+    acquired = retry_quota.acquire(error=Exception())
 
     assert acquired == 3
     assert retry_quota.available_capacity == 7
 
 
-async def test_retry_quota_acquire_when_exhausted(
+def test_retry_quota_acquire_when_exhausted(
     retry_quota: StandardRetryQuota,
 ) -> None:
     # Drain capacity: 10 -> 7 -> 4 -> 1
-    await retry_quota.acquire(error=Exception())
-    await retry_quota.acquire(error=Exception())
-    await retry_quota.acquire(error=Exception())
+    retry_quota.acquire(error=Exception())
+    retry_quota.acquire(error=Exception())
+    retry_quota.acquire(error=Exception())
     assert retry_quota.available_capacity == 1
 
     # Next acquire needs 3 but only 1 remains
     with pytest.raises(RetryError, match="Retry quota exceeded"):
-        await retry_quota.acquire(error=Exception())
+        retry_quota.acquire(error=Exception())
 
 
-async def test_retry_quota_release_restores_capacity(
+def test_retry_quota_release_restores_capacity(
     retry_quota: StandardRetryQuota,
 ) -> None:
-    acquired = await retry_quota.acquire(error=Exception())
+    acquired = retry_quota.acquire(error=Exception())
     assert retry_quota.available_capacity == 7
 
-    await retry_quota.release(release_amount=acquired)
+    retry_quota.release(release_amount=acquired)
     assert retry_quota.available_capacity == 10
 
 
-async def test_retry_quota_release_zero_adds_increment(
+def test_retry_quota_release_zero_adds_increment(
     retry_quota: StandardRetryQuota,
 ) -> None:
-    await retry_quota.acquire(error=Exception())
+    retry_quota.acquire(error=Exception())
     assert retry_quota.available_capacity == 7
 
-    await retry_quota.release(release_amount=0)
+    retry_quota.release(release_amount=0)
     assert retry_quota.available_capacity == 8
 
 
-async def test_retry_quota_release_caps_at_max(
+def test_retry_quota_release_caps_at_max(
     retry_quota: StandardRetryQuota,
 ) -> None:
     # Drain some capacity
-    await retry_quota.acquire(error=Exception())
-    await retry_quota.acquire(error=Exception())
+    retry_quota.acquire(error=Exception())
+    retry_quota.acquire(error=Exception())
     assert retry_quota.available_capacity == 4
 
     # Release more than drained. Should cap at max
-    await retry_quota.release(release_amount=20)
+    retry_quota.release(release_amount=20)
     assert retry_quota.available_capacity == 10

From d96c0dbc8d3a39af464d32a7abbc79adcff1a7a0 Mon Sep 17 00:00:00 2001
From: Alessandra Romero <romero.alessandra1@gmail.com>
Date: Thu, 13 Nov 2025 12:18:30 -0500
Subject: [PATCH 09/11] Parametrize retries unit tests

---
 .../smithy-core/tests/unit/test_retries.py    | 31 +++++++++----------
 1 file changed, 14 insertions(+), 17 deletions(-)

diff --git a/packages/smithy-core/tests/unit/test_retries.py b/packages/smithy-core/tests/unit/test_retries.py
index ece1a394f..1983393c3 100644
--- a/packages/smithy-core/tests/unit/test_retries.py
+++ b/packages/smithy-core/tests/unit/test_retries.py
@@ -120,24 +120,21 @@ def test_standard_retry_strategy(max_attempts: int) -> None:
         strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
 
 
-def test_standard_retry_does_not_retry_unclassified() -> None:
-    strategy = StandardRetryStrategy()
-    token = strategy.acquire_initial_retry_token()
-    with pytest.raises(RetryError):
-        strategy.refresh_retry_token_for_retry(token_to_renew=token, error=Exception())
-
-
-def test_standard_retry_does_not_retry_when_safety_unknown() -> None:
-    strategy = StandardRetryStrategy()
-    error = CallError(is_retry_safe=None)
-    token = strategy.acquire_initial_retry_token()
-    with pytest.raises(RetryError):
-        strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
-
-
-def test_standard_retry_does_not_retry_unsafe() -> None:
+@pytest.mark.parametrize(
+    "error",
+    [
+        Exception(),
+        CallError(is_retry_safe=None),
+        CallError(fault="client", is_retry_safe=False),
+    ],
+    ids=[
+        "unclassified_error",
+        "safety_unknown_error",
+        "unsafe_error",
+    ],
+)
+def test_standard_retry_does_not_retry(error: Exception | CallError) -> None:
     strategy = StandardRetryStrategy()
-    error = CallError(fault="client", is_retry_safe=False)
     token = strategy.acquire_initial_retry_token()
     with pytest.raises(RetryError):
         strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)

From 21cde20d0eb855ba399c30276be58f687f9f4f6c Mon Sep 17 00:00:00 2001
From: Alessandra Romero <romero.alessandra1@gmail.com>
Date: Thu, 13 Nov 2025 14:24:05 -0500
Subject: [PATCH 10/11] Update StandardRetryQuota class and corresponding unit
 tests

---
 .../smithy-core/src/smithy_core/retries.py    | 110 +++++++++---------
 .../smithy-core/tests/unit/test_retries.py    |  64 ++--------
 2 files changed, 67 insertions(+), 107 deletions(-)

diff --git a/packages/smithy-core/src/smithy_core/retries.py b/packages/smithy-core/src/smithy_core/retries.py
index ccc2bbbbb..b56cc87c2 100644
--- a/packages/smithy-core/src/smithy_core/retries.py
+++ b/packages/smithy-core/src/smithy_core/retries.py
@@ -245,6 +245,59 @@ def record_success(self, *, token: retries_interface.RetryToken) -> None:
         """Not used by this retry strategy."""
 
 
+class StandardRetryQuota:
+    """Retry quota used by :py:class:`StandardRetryStrategy`."""
+
+    INITIAL_RETRY_TOKENS: int = 500
+    RETRY_COST: int = 5
+    NO_RETRY_INCREMENT: int = 1
+    TIMEOUT_RETRY_COST: int = 10
+
+    def __init__(self, initial_capacity: int = INITIAL_RETRY_TOKENS):
+        """Initialize retry quota with configurable capacity.
+
+        :param initial_capacity: The initial and maximum capacity for the retry quota.
+        """
+        self._max_capacity = initial_capacity
+        self._available_capacity = initial_capacity
+        self._lock = threading.Lock()
+
+    def acquire(self, *, error: Exception) -> int:
+        """Attempt to acquire capacity for a retry attempt.
+
+        If there's insufficient capacity available, raise an exception.
+        Otherwise, return the amount of capacity successfully allocated.
+        """
+        capacity_amount = self.RETRY_COST
+
+        with self._lock:
+            if capacity_amount > self._available_capacity:
+                raise RetryError("Retry quota exceeded")
+            self._available_capacity -= capacity_amount
+            return capacity_amount
+
+    def release(self, *, release_amount: int) -> None:
+        """Release capacity back to the retry quota.
+
+        The capacity being released will be truncated if necessary to ensure the max
+        capacity is never exceeded.
+        """
+        increment = self.NO_RETRY_INCREMENT if release_amount == 0 else release_amount
+
+        if self._available_capacity == self._max_capacity:
+            return
+
+        with self._lock:
+            self._available_capacity = min(
+                self._available_capacity + increment, self._max_capacity
+            )
+
+    @property
+    def available_capacity(self) -> int:
+        """Return the amount of capacity available."""
+        return self._available_capacity
+
+
 @dataclass(kw_only=True)
 class StandardRetryToken:
     retry_count: int
@@ -266,6 +319,7 @@ def __init__(
         *,
         backoff_strategy: retries_interface.RetryBackoffStrategy | None = None,
         max_attempts: int = 3,
+        retry_quota: StandardRetryQuota | None = None,
     ):
         """Standard retry strategy using truncated binary exponential backoff with full
         jitter.
@@ -275,6 +329,9 @@ def __init__(
 
         :param max_attempts: Upper limit on total number of attempts made, including
             initial attempt and retries.
+
+        :param retry_quota: The retry quota to use for managing retry capacity. Defaults
+            to a new :py:class:`StandardRetryQuota` instance.
         """
         if max_attempts < 0:
             raise ValueError(
@@ -287,7 +344,7 @@ def __init__(
             jitter_type=ExponentialBackoffJitterType.FULL,
         )
         self.max_attempts = max_attempts
-        self._retry_quota = StandardRetryQuota()
+        self._retry_quota = retry_quota or StandardRetryQuota()
 
     def acquire_initial_retry_token(
         self, *, token_scope: str | None = None
@@ -357,54 +414,3 @@ def record_success(self, *, token: retries_interface.RetryToken) -> None:
                 f"StandardRetryStrategy requires StandardRetryToken, got {type(token).__name__}"
             )
         self._retry_quota.release(release_amount=token.last_quota_acquired)
-
-
-class StandardRetryQuota:
-    """Retry quota used by :py:class:`StandardRetryStrategy`."""
-
-    INITIAL_RETRY_TOKENS: int = 500
-    RETRY_COST: int = 5
-    NO_RETRY_INCREMENT: int = 1
-    TIMEOUT_RETRY_COST: int = 10
-
-    def __init__(self):
-        self._max_capacity = self.INITIAL_RETRY_TOKENS
-        self._available_capacity = self.INITIAL_RETRY_TOKENS
-        self._lock = threading.Lock()
-
-    def acquire(self, *, error: Exception) -> int:
-        """Attempt to acquire a certain amount of capacity.
-
-        If there's insufficient capacity available, raise an exception.
-        Otherwise, we return the amount of capacity successfully allocated.
-        """
-        # TODO: update `is_timeout` when `is_timeout_error` is implemented
-        is_timeout = False
-        capacity_amount = self.TIMEOUT_RETRY_COST if is_timeout else self.RETRY_COST
-
-        with self._lock:
-            if capacity_amount > self._available_capacity:
-                raise RetryError("Retry quota exceeded")
-            self._available_capacity -= capacity_amount
-            return capacity_amount
-
-    def release(self, *, release_amount: int) -> None:
-        """Release capacity back to the retry quota.
-
-        The capacity being released will be truncated if necessary to ensure the max
-        capacity is never exceeded.
-        """
-        increment = self.NO_RETRY_INCREMENT if release_amount == 0 else release_amount
-
-        if self._available_capacity == self._max_capacity:
-            return
-
-        with self._lock:
-            self._available_capacity = min(
-                self._available_capacity + increment, self._max_capacity
-            )
-
-    @property
-    def available_capacity(self) -> int:
-        """Return the amount of capacity available."""
-        return self._available_capacity
diff --git a/packages/smithy-core/tests/unit/test_retries.py b/packages/smithy-core/tests/unit/test_retries.py
index 1983393c3..63b257050 100644
--- a/packages/smithy-core/tests/unit/test_retries.py
+++ b/packages/smithy-core/tests/unit/test_retries.py
@@ -166,72 +166,32 @@ def test_standard_retry_invalid_max_attempts() -> None:
         StandardRetryStrategy(max_attempts=-1)
 
 
-def test_standard_retry_record_success_without_retry() -> None:
-    strategy = StandardRetryStrategy()
-    token = strategy.acquire_initial_retry_token()
-    initial_capacity = strategy._retry_quota.available_capacity  # pyright: ignore[reportPrivateUsage]
-
-    strategy.record_success(token=token)
-
-    # Should increment by NO_RETRY_INCREMENT
-    expected = min(
-        initial_capacity + StandardRetryQuota.NO_RETRY_INCREMENT,
-        StandardRetryQuota.INITIAL_RETRY_TOKENS,
-    )
-    assert strategy._retry_quota.available_capacity == expected  # pyright: ignore[reportPrivateUsage]
-
-
-def test_standard_retry_record_success_with_retry() -> None:
-    strategy = StandardRetryStrategy()
-    error = CallError(is_retry_safe=True)
-    token = strategy.acquire_initial_retry_token()
-
-    token = strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
-    capacity_after_retry = strategy._retry_quota.available_capacity  # pyright: ignore[reportPrivateUsage]
-
-    strategy.record_success(token=token)
-
-    # Capacity should increase by last_quota_acquired
-    assert (
-        strategy._retry_quota.available_capacity  # pyright: ignore[reportPrivateUsage]
-        == capacity_after_retry + token.last_quota_acquired
-    )
-
-
 @pytest.fixture
-def retry_quota(monkeypatch: pytest.MonkeyPatch) -> StandardRetryQuota:
-    monkeypatch.setattr(StandardRetryQuota, "INITIAL_RETRY_TOKENS", 10, raising=False)
-    monkeypatch.setattr(StandardRetryQuota, "RETRY_COST", 3, raising=False)
-    monkeypatch.setattr(StandardRetryQuota, "NO_RETRY_INCREMENT", 1, raising=False)
-    return StandardRetryQuota()
+def retry_quota() -> StandardRetryQuota:
+    return StandardRetryQuota(initial_capacity=10)
 
 
 def test_retry_quota_initial_state(
     retry_quota: StandardRetryQuota,
 ) -> None:
     assert retry_quota.available_capacity == 10
-    assert retry_quota._max_capacity == 10  # pyright: ignore[reportPrivateUsage]
 
 
 def test_retry_quota_acquire_success(
     retry_quota: StandardRetryQuota,
 ) -> None:
     acquired = retry_quota.acquire(error=Exception())
-
-    assert acquired == 3
-    assert retry_quota.available_capacity == 7
+    assert retry_quota.available_capacity == 10 - acquired
 
 
 def test_retry_quota_acquire_when_exhausted(
     retry_quota: StandardRetryQuota,
 ) -> None:
-    # Drain capacity: 10 -> 7 -> 4 -> 1
+    # Drain capacity until insufficient for next acquire
     retry_quota.acquire(error=Exception())
     retry_quota.acquire(error=Exception())
-    retry_quota.acquire(error=Exception())
-    assert retry_quota.available_capacity == 1
 
-    # Next acquire needs 3 but only 1 remains
+    # Not enough capacity for another retry (need 5, only 0 left)
     with pytest.raises(RetryError, match="Retry quota exceeded"):
         retry_quota.acquire(error=Exception())
 
@@ -240,8 +200,6 @@ def test_retry_quota_release_restores_capacity(
     retry_quota: StandardRetryQuota,
 ) -> None:
     acquired = retry_quota.acquire(error=Exception())
-    assert retry_quota.available_capacity == 7
-
     retry_quota.release(release_amount=acquired)
     assert retry_quota.available_capacity == 10
 
@@ -250,10 +208,9 @@ def test_retry_quota_release_zero_adds_increment(
     retry_quota: StandardRetryQuota,
 ) -> None:
     retry_quota.acquire(error=Exception())
-    assert retry_quota.available_capacity == 7
-
+    assert retry_quota.available_capacity == 5
     retry_quota.release(release_amount=0)
-    assert retry_quota.available_capacity == 8
+    assert retry_quota.available_capacity == 6
 
 
 def test_retry_quota_release_caps_at_max(
@@ -261,9 +218,6 @@ def test_retry_quota_release_caps_at_max(
 ) -> None:
     # Drain some capacity
     retry_quota.acquire(error=Exception())
-    retry_quota.acquire(error=Exception())
-    assert retry_quota.available_capacity == 4
-
-    # Release more than drained. Should cap at max
-    retry_quota.release(release_amount=20)
+    # Release more than we acquired. Should cap at initial capacity.
+    retry_quota.release(release_amount=50)
     assert retry_quota.available_capacity == 10

From d1de172475a6f278e913953bd99708ee23f7a920 Mon Sep 17 00:00:00 2001
From: Alessandra Romero <romero.alessandra1@gmail.com>
Date: Mon, 17 Nov 2025 11:56:24 -0500
Subject: [PATCH 11/11] Remove unused quota_consumed field from
 StandardRetryToken

---
 packages/smithy-core/src/smithy_core/retries.py | 13 ++++---------
 packages/smithy-core/tests/unit/test_retries.py | 13 -------------
 2 files changed, 4 insertions(+), 22 deletions(-)

diff --git a/packages/smithy-core/src/smithy_core/retries.py b/packages/smithy-core/src/smithy_core/retries.py
index b56cc87c2..0f017f6e2 100644
--- a/packages/smithy-core/src/smithy_core/retries.py
+++ b/packages/smithy-core/src/smithy_core/retries.py
@@ -306,11 +306,8 @@ class StandardRetryToken:
     retry_delay: float
     """Delay in seconds to wait before the retry attempt."""
 
-    quota_consumed: int = 0
-    """The total amount of quota consumed."""
-
-    last_quota_acquired: int = 0
-    """The amount of last quota acquired."""
+    quota_acquired: int = 0
+    """The amount of quota acquired for this retry attempt."""
 
 
 class StandardRetryStrategy(retries_interface.RetryStrategy):
@@ -386,7 +383,6 @@ def refresh_retry_token_for_retry(
             # Acquire additional quota for this retry attempt
             # (may raise a RetryError if none is available)
             quota_acquired = self._retry_quota.acquire(error=error)
-            total_quota: int = token_to_renew.quota_consumed + quota_acquired
 
             if error.retry_after is not None:
                 retry_delay = error.retry_after
@@ -398,8 +394,7 @@ def refresh_retry_token_for_retry(
             return StandardRetryToken(
                 retry_count=retry_count,
                 retry_delay=retry_delay,
-                quota_consumed=total_quota,
-                last_quota_acquired=quota_acquired,
+                quota_acquired=quota_acquired,
             )
         else:
             raise RetryError(f"Error is not retryable: {error}") from error
@@ -413,4 +408,4 @@ def record_success(self, *, token: retries_interface.RetryToken) -> None:
             raise TypeError(
                 f"StandardRetryStrategy requires StandardRetryToken, got {type(token).__name__}"
             )
-        self._retry_quota.release(release_amount=token.last_quota_acquired)
+        self._retry_quota.release(release_amount=token.quota_acquired)
diff --git a/packages/smithy-core/tests/unit/test_retries.py b/packages/smithy-core/tests/unit/test_retries.py
index 63b257050..f8abea721 100644
--- a/packages/smithy-core/tests/unit/test_retries.py
+++ b/packages/smithy-core/tests/unit/test_retries.py
@@ -148,19 +148,6 @@ def test_standard_retry_after_overrides_backoff() -> None:
     assert token.retry_delay == 5.5
 
 
-def test_standard_retry_quota_consumed_accumulates() -> None:
-    strategy = StandardRetryStrategy()
-    error = CallError(is_retry_safe=True)
-    token = strategy.acquire_initial_retry_token()
-
-    token = strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
-    first_consumed = token.quota_consumed
-    assert first_consumed == StandardRetryQuota.RETRY_COST
-
-    token = strategy.refresh_retry_token_for_retry(token_to_renew=token, error=error)
-    assert token.quota_consumed == first_consumed + StandardRetryQuota.RETRY_COST
-
-
 def test_standard_retry_invalid_max_attempts() -> None:
     with pytest.raises(ValueError):
         StandardRetryStrategy(max_attempts=-1)