diff --git a/SPEC.md b/SPEC.md
index 0916f9de3..2b1808142 100644
--- a/SPEC.md
+++ b/SPEC.md
@@ -255,7 +255,7 @@ END
 ### 401 Refresh-and-Retry Algorithm
 
 1. Receive 401 response.
-2. If the token provider supports refresh (`refreshable() == true`) and refresh has not yet been attempted for this request:
+2. If the token provider supports refresh (`refreshable() == true`), refresh has not yet been attempted for this request, **and the attempt budget has another attempt left**:
    a. Call `refresh()`.
    b. If refresh succeeded, retry the request once with updated token.
    c. → response from retry.
@@ -263,6 +263,28 @@ END
 
 Refresh is attempted at most once per request. Implementations track this with a boolean (e.g., `refresh_attempted`) rather than a counter.
 
+**The replay spends an attempt.** It puts a request on the wire, so it draws
+from the same total-attempt budget as a transient retry (§7): `max_retries`
+counts requests, not failure kinds, and a cap of one means one request no
+matter what would have caused the second. That is what makes §14's "disabling
+retry yields exactly ONE hop-1 attempt" literally true — refresh included —
+and it preserves the total-attempt semantics settled for observed attempts in
+#461.
+
+Step 2's budget gate is checked **before** `refresh()` is called, not after.
+Refreshing a token the SDK has no budget left to use would burn a rotation for
+nothing and still hand the caller the stale 401; declining to refresh at all is
+both cheaper and easier to reason about. The consequence is worth stating
+plainly: with a budget of one attempt, a refreshable 401 is NOT replayed and
+surfaces as `auth_required`. Callers who want the refresh replay must leave an
+attempt for it.
+
+This gate applies wherever a total-attempt budget governs the path. Go's
+hand-written mutation path has no such budget — mutations are deliberately
+single-attempt for transient failures — and keeps its documented
+mutation-specific single re-attempt after a successful refresh (see §7's
+Cross-SDK Divergence).
+
 ---
 
 ## §5. Service Surface
diff --git a/python/src/basecamp/_async_http.py b/python/src/basecamp/_async_http.py
index f6b4e2cdd..8fd7acc88 100644
--- a/python/src/basecamp/_async_http.py
+++ b/python/src/basecamp/_async_http.py
@@ -163,6 +163,7 @@ async def _request_with_retry(
         max_attempts = self._config.max_retries if self._config.max_retries > 0 else 1
         max_attempts = self._apply_operation_retry_max(operation, max_attempts)
         attempt = 0
+        refreshed_once = False
         last_error: BasecampError | None = None
 
         while True:
@@ -170,6 +171,13 @@ async def _request_with_retry(
             if attempt > max_attempts:
                 break
 
+            # Each except suite only CLASSIFIES the attempt; the retry side
+            # effects run below, outside every handler. An exception raised
+            # inside an except suite bypasses that try's sibling handlers, so
+            # refreshing there would let a NetworkError from the token endpoint
+            # escape the loop with budget still on the table.
+            error: BasecampError | None = None
+            stale_auth: AuthError | None = None
             try:
                 return await self._single_request(
                     method,
@@ -182,19 +190,63 @@ async def _request_with_retry(
                     attempt=attempt,
                     allow_cross_origin=allow_cross_origin,
                     accept=accept,
+                    refresh_replay=False,
                 )
-            except (RateLimitError, NetworkError, ApiError) as e:
-                if not self._is_retryable_error(e, operation, retry_on=retry_on):
+            except AuthError as e:
+                # SPEC §4: the refresh replay is a request on the wire, so it
+                # spends an attempt from THIS budget rather than an uncounted
+                # one inside the single-request primitive. max_retries is a
+                # total attempt count (#461), and a cap of one means one
+                # request whatever would have caused the second.
+                #
+                # The budget is checked BEFORE refresh() so a rotation is never
+                # burned on an attempt the loop has no room to make. Await the
+                # refresh rather than truthy-testing the coroutine, which would
+                # pass the gate without ever rotating the token.
+                if refreshed_once or e.http_status != 401 or attempt >= max_attempts:
                     raise
-                last_error = e
+                stale_auth = e
+            except (RateLimitError, NetworkError, ApiError) as e:
+                error = e
+
+            if stale_auth is not None:
+                # SPEC §4 tracks refresh with an "attempted" boolean, not a
+                # "succeeded" one, so mark it BEFORE invoking the provider: a
+                # refresh that throws still counts as the one attempt this
+                # request gets. Otherwise a transient token-endpoint failure
+                # lets the NEXT 401 in the same request call refresh() again —
+                # and if the first call reached the server and rotated the
+                # token before its response was lost, the second spends a
+                # refresh token that is already dead.
+                refreshed_once = True
+                try:
+                    tp = getattr(self._auth, "token_provider", None)
+                    refreshed = bool(tp and getattr(tp, "refreshable", False) and await tp.refresh())
+                except (RateLimitError, NetworkError, ApiError) as e:
+                    # A token endpoint that times out is a transient failure of
+                    # this attempt, not a terminal auth fault: it retries under
+                    # the same budget, exactly as it did when the replay lived
+                    # inside _single_request.
+                    error = e
+                else:
+                    if not refreshed:
+                        raise stale_auth
+                    # No backoff: the token is fresh, and the server never asked
+                    # us to wait. The replay still costs the attempt counted above.
+                    continue
+
+            if error is not None:
+                if not self._is_retryable_error(error, operation, retry_on=retry_on):
+                    raise error
+                last_error = error
                 if attempt >= max_attempts:
                     break
-                delay = self._calculate_delay(attempt, e.retry_after)
+                delay = self._calculate_delay(attempt, error.retry_after)
                 safe_hook(
                     self._hooks.on_retry,
                     RequestInfo(method=method, url=url, attempt=attempt),
                     attempt + 1,
-                    e,
+                    error,
                     delay,
                 )
                 await asyncio.sleep(delay)
@@ -218,6 +270,7 @@ async def _single_request(
         _retry_count: int = 0,
         allow_cross_origin: bool = False,
         accept: str | None = "application/json",
+        refresh_replay: bool = True,
     ) -> httpx.Response:
         if not allow_cross_origin and not (
             _security.is_localhost(url) or _security.same_origin(url, self._config.base_url)
@@ -247,8 +300,11 @@ async def _single_request(
 
             if response.status_code >= 400:
                 error = self._handle_error(response)
-                # 401 retry with async token refresh
-                if isinstance(error, AuthError) and error.http_status == 401 and _retry_count < 1:
+                # 401 replay for callers that come here directly (mutations),
+                # which have no retry loop to own it. _request_with_retry
+                # passes refresh_replay=False and replays from the loop so the
+                # extra request draws from the attempt budget (SPEC §4).
+                if refresh_replay and isinstance(error, AuthError) and error.http_status == 401 and _retry_count < 1:
                     tp = getattr(self._auth, "token_provider", None)
                     if tp and getattr(tp, "refreshable", False) and await tp.refresh():
                         return await self._single_request(
@@ -263,6 +319,7 @@ async def _single_request(
                             _retry_count=_retry_count + 1,
                             allow_cross_origin=allow_cross_origin,
                             accept=accept,
+                            refresh_replay=refresh_replay,
                         )
                 raise error
 
diff --git a/python/src/basecamp/_http.py b/python/src/basecamp/_http.py
index abf9b2be3..5a6f63de4 100644
--- a/python/src/basecamp/_http.py
+++ b/python/src/basecamp/_http.py
@@ -163,6 +163,7 @@ def _request_with_retry(
         max_attempts = self._config.max_retries if self._config.max_retries > 0 else 1
         max_attempts = self._apply_operation_retry_max(operation, max_attempts)
         attempt = 0
+        refreshed_once = False
         last_error: BasecampError | None = None
 
         while True:
@@ -170,6 +171,13 @@ def _request_with_retry(
             if attempt > max_attempts:
                 break
 
+            # Each except suite only CLASSIFIES the attempt; the retry side
+            # effects run below, outside every handler. An exception raised
+            # inside an except suite bypasses that try's sibling handlers, so
+            # refreshing there would let a NetworkError from the token endpoint
+            # escape the loop with budget still on the table.
+            error: BasecampError | None = None
+            stale_auth: AuthError | None = None
             try:
                 return self._single_request(
                     method,
@@ -182,19 +190,61 @@ def _request_with_retry(
                     attempt=attempt,
                     allow_cross_origin=allow_cross_origin,
                     accept=accept,
+                    refresh_replay=False,
                 )
-            except (RateLimitError, NetworkError, ApiError) as e:
-                if not self._is_retryable_error(e, operation, retry_on=retry_on):
+            except AuthError as e:
+                # SPEC §4: the refresh replay is a request on the wire, so it
+                # spends an attempt from THIS budget rather than an uncounted
+                # one inside the single-request primitive. max_retries is a
+                # total attempt count (#461), and a cap of one means one
+                # request whatever would have caused the second.
+                #
+                # The budget is checked BEFORE refresh() so a rotation is never
+                # burned on an attempt the loop has no room to make.
+                if refreshed_once or e.http_status != 401 or attempt >= max_attempts:
                     raise
-                last_error = e
+                stale_auth = e
+            except (RateLimitError, NetworkError, ApiError) as e:
+                error = e
+
+            if stale_auth is not None:
+                # SPEC §4 tracks refresh with an "attempted" boolean, not a
+                # "succeeded" one, so mark it BEFORE invoking the provider: a
+                # refresh that throws still counts as the one attempt this
+                # request gets. Otherwise a transient token-endpoint failure
+                # lets the NEXT 401 in the same request call refresh() again —
+                # and if the first call reached the server and rotated the
+                # token before its response was lost, the second spends a
+                # refresh token that is already dead.
+                refreshed_once = True
+                try:
+                    tp = getattr(self._auth, "token_provider", None)
+                    refreshed = bool(tp and getattr(tp, "refreshable", False) and tp.refresh())
+                except (RateLimitError, NetworkError, ApiError) as e:
+                    # A token endpoint that times out is a transient failure of
+                    # this attempt, not a terminal auth fault: it retries under
+                    # the same budget, exactly as it did when the replay lived
+                    # inside _single_request.
+                    error = e
+                else:
+                    if not refreshed:
+                        raise stale_auth
+                    # No backoff: the token is fresh, and the server never asked
+                    # us to wait. The replay still costs the attempt counted above.
+                    continue
+
+            if error is not None:
+                if not self._is_retryable_error(error, operation, retry_on=retry_on):
+                    raise error
+                last_error = error
                 if attempt >= max_attempts:
                     break
-                delay = self._calculate_delay(attempt, e.retry_after)
+                delay = self._calculate_delay(attempt, error.retry_after)
                 safe_hook(
                     self._hooks.on_retry,
                     RequestInfo(method=method, url=url, attempt=attempt),
                     attempt + 1,
-                    e,
+                    error,
                     delay,
                 )
                 time.sleep(delay)
@@ -218,6 +268,7 @@ def _single_request(
         _retry_count: int = 0,
         allow_cross_origin: bool = False,
         accept: str | None = "application/json",
+        refresh_replay: bool = True,
     ) -> httpx.Response:
         if not allow_cross_origin and not (
             _security.is_localhost(url) or _security.same_origin(url, self._config.base_url)
@@ -247,8 +298,11 @@ def _single_request(
 
             if response.status_code >= 400:
                 error = self._handle_error(response)
-                # 401 retry with token refresh
-                if isinstance(error, AuthError) and error.http_status == 401 and _retry_count < 1:
+                # 401 replay for callers that come here directly (mutations),
+                # which have no retry loop to own it. _request_with_retry
+                # passes refresh_replay=False and replays from the loop so the
+                # extra request draws from the attempt budget (SPEC §4).
+                if refresh_replay and isinstance(error, AuthError) and error.http_status == 401 and _retry_count < 1:
                     tp = getattr(self._auth, "token_provider", None)
                     if tp and getattr(tp, "refreshable", False) and tp.refresh():
                         return self._single_request(
@@ -263,6 +317,7 @@ def _single_request(
                             _retry_count=_retry_count + 1,
                             allow_cross_origin=allow_cross_origin,
                             accept=accept,
+                            refresh_replay=refresh_replay,
                         )
                 raise error
 
diff --git a/python/tests/test_download.py b/python/tests/test_download.py
index f96e5154d..55763b102 100644
--- a/python/tests/test_download.py
+++ b/python/tests/test_download.py
@@ -12,10 +12,36 @@
 from basecamp.auth import BearerAuth, StaticTokenProvider
 from basecamp.config import Config
 from basecamp.download import _rewrite_url, download_async, download_sync, filename_from_url
-from basecamp.errors import ApiError, BasecampError, UsageError
+from basecamp.errors import ApiError, AuthError, UsageError
 from basecamp.hooks import BasecampHooks
 
 
+class _RefreshableProvider(StaticTokenProvider):
+    """Counts refreshes so a test can tell "declined to refresh" from "refreshed"."""
+
+    refreshable = True
+
+    def __init__(self):
+        super().__init__("test-token")
+        self.refreshes = 0
+
+    def refresh(self):
+        self.refreshes += 1
+        return True
+
+
+class _AsyncRefreshableProvider(AsyncStaticTokenProvider):
+    refreshable = True
+
+    def __init__(self):
+        super().__init__("test-token")
+        self.refreshes = 0
+
+    async def refresh(self):
+        self.refreshes += 1
+        return True
+
+
 def make_config():
     return Config(base_url="https://3.basecampapi.com")
 
@@ -236,40 +262,67 @@ def test_auth_on_every_hop1_attempt_never_on_hop2(self):
         assert "Authorization" not in hop2.calls[0].request.headers
 
     @respx.mock
-    def test_refreshable_401_replays_once_independent_of_the_retry_cap(self):
-        """SPEC §4: refresh is attempted at most once PER REQUEST.
-
-        §4 tracks it with a boolean, not a counter, and does not subordinate
-        it to the transient-retry budget — so a refreshable 401 replays once
-        even with retries disabled. That is deliberately orthogonal to §14's
-        hop-1 attempt cap, which governs transient-failure retries. Whether
-        the two should be reconciled is tracked in #565; this pins what the
-        spec says today so a future change has to be a decision, not a drift.
-        """
+    def test_refresh_replay_is_not_attempted_without_budget(self):
+        """SPEC §4/§14: with retry disabled, ONE request goes out — refresh included.
 
-        class RefreshableProvider(StaticTokenProvider):
-            refreshable = True
+        The replay is a request on the wire, so it spends an attempt from the
+        same total-attempt budget as a transient retry. A cap of one means one
+        request no matter what would have caused the second (#565, #461).
 
-            def __init__(self):
-                super().__init__("test-token")
-                self.refreshes = 0
+        The refresh itself must not fire either: §4 checks the budget BEFORE
+        calling refresh(), because rotating a token the SDK has no attempt left
+        to use burns it for nothing and still surfaces the stale 401.
+        """
+        hop1 = respx.get(self.HOP1).mock(return_value=httpx.Response(401))
+
+        provider = _RefreshableProvider()
+        config = make_fast_config(max_retries=0)
+        http = HttpClient(config, BearerAuth(provider))
+        with pytest.raises(AuthError):
+            download_sync(self.RAW, http_client=http, config=config)
 
-            def refresh(self):
-                self.refreshes += 1
-                return True
+        assert hop1.call_count == 1
+        assert provider.refreshes == 0
 
+    @respx.mock
+    def test_refresh_replay_spends_an_attempt_from_the_budget(self):
+        """A budget of two allows the refresh replay, and it consumes attempt 2."""
         hop1 = respx.get(self.HOP1).mock(return_value=httpx.Response(401))
 
-        provider = RefreshableProvider()
-        config = make_fast_config(max_retries=0)
+        provider = _RefreshableProvider()
+        config = make_fast_config(max_retries=2)
         http = HttpClient(config, BearerAuth(provider))
-        with pytest.raises(BasecampError):
+        with pytest.raises(AuthError):
             download_sync(self.RAW, http_client=http, config=config)
 
-        # One transient attempt, plus §4's single refresh replay.
+        # Two requests, not three: the replay drew from the same budget, and
+        # §4 allows only one refresh per request regardless.
         assert hop1.call_count == 2
         assert provider.refreshes == 1
 
+    @respx.mock
+    def test_refresh_replay_succeeds_when_the_new_token_works(self):
+        """The refreshed request is actually SENT, not refreshed-then-discarded."""
+        hop1 = respx.get(self.HOP1).mock(
+            side_effect=[
+                httpx.Response(401),
+                httpx.Response(302, headers={"Location": self.SIGNED}),
+            ]
+        )
+        hop2 = respx.get(self.SIGNED).mock(
+            return_value=httpx.Response(200, content=b"data", headers={"content-type": "application/pdf"})
+        )
+
+        provider = _RefreshableProvider()
+        config = make_fast_config(max_retries=2)
+        http = HttpClient(config, BearerAuth(provider))
+        result = download_sync(self.RAW, http_client=http, config=config)
+
+        assert result.body == b"data"
+        assert hop1.call_count == 2
+        assert hop2.call_count == 1
+        assert provider.refreshes == 1
+
     @respx.mock
     def test_hop1_sends_no_accept_header_on_any_attempt(self):
         """SPEC section 14: hop 1 carries Authorization and User-Agent only.
@@ -478,3 +531,60 @@ async def test_hop1_sends_no_accept_header_on_any_attempt(self):
             assert call.request.headers.get("Accept") != "application/json"
             assert "Content-Type" not in call.request.headers
             assert call.request.headers.get("User-Agent") is not None
+
+    @respx.mock
+    @pytest.mark.asyncio
+    async def test_refresh_replay_is_not_attempted_without_budget(self):
+        """SPEC §4/§14 budget gate, async transport (see the sync twin)."""
+        hop1 = respx.get(self.HOP1).mock(return_value=httpx.Response(401))
+
+        provider = _AsyncRefreshableProvider()
+        config = make_fast_config(max_retries=0)
+        http = AsyncHttpClient(config, AsyncBearerAuth(provider))
+        with pytest.raises(AuthError):
+            await download_async(self.RAW, http_client=http, config=config)
+
+        assert hop1.call_count == 1
+        assert provider.refreshes == 0
+
+    @respx.mock
+    @pytest.mark.asyncio
+    async def test_refresh_replay_spends_an_attempt_from_the_budget(self):
+        hop1 = respx.get(self.HOP1).mock(return_value=httpx.Response(401))
+
+        provider = _AsyncRefreshableProvider()
+        config = make_fast_config(max_retries=2)
+        http = AsyncHttpClient(config, AsyncBearerAuth(provider))
+        with pytest.raises(AuthError):
+            await download_async(self.RAW, http_client=http, config=config)
+
+        assert hop1.call_count == 2
+        assert provider.refreshes == 1
+
+    @respx.mock
+    @pytest.mark.asyncio
+    async def test_refresh_replay_succeeds_when_the_new_token_works(self):
+        """The refreshed request is actually SENT, not awaited-then-discarded.
+
+        Pins the await too: a refresh coroutine truthy-tested without awaiting
+        would pass the gate and then never rotate the token (#563's de8c9255d).
+        """
+        hop1 = respx.get(self.HOP1).mock(
+            side_effect=[
+                httpx.Response(401),
+                httpx.Response(302, headers={"Location": self.SIGNED}),
+            ]
+        )
+        hop2 = respx.get(self.SIGNED).mock(
+            return_value=httpx.Response(200, content=b"data", headers={"content-type": "application/pdf"})
+        )
+
+        provider = _AsyncRefreshableProvider()
+        config = make_fast_config(max_retries=2)
+        http = AsyncHttpClient(config, AsyncBearerAuth(provider))
+        result = await download_async(self.RAW, http_client=http, config=config)
+
+        assert result.body == b"data"
+        assert hop1.call_count == 2
+        assert hop2.call_count == 1
+        assert provider.refreshes == 1
diff --git a/python/tests/test_http.py b/python/tests/test_http.py
index 1dde94ecf..01050aef0 100644
--- a/python/tests/test_http.py
+++ b/python/tests/test_http.py
@@ -697,3 +697,154 @@ async def test_async_explicitly_empty_retry_on_never_retries(self):
         with pytest.raises(ApiError):
             await client.put("/thing", json_body={"x": 1}, operation="OptOutOp")
         assert route.call_count == 1
+
+
+class TestRefreshReplayAttemptBudget:
+    """SPEC §4: the 401 refresh replay draws from the total-attempt budget.
+
+    These pin the PLAIN GET path, not just the download hop — the "every GET
+    in Python and Ruby" case that the earlier, ungated attempt at this change
+    regressed (#563's aca2c481b, now superseded by §4's budget gate). #565.
+    """
+
+    URL = "https://3.basecampapi.com/thing"
+
+    def make_client(self, max_retries):
+        config = Config(
+            base_url="https://3.basecampapi.com",
+            max_retries=max_retries,
+            base_delay=0.001,
+            max_jitter=0.0,
+        )
+        provider = _RefreshableProvider()
+        return HttpClient(config, BearerAuth(provider), BasecampHooks()), provider
+
+    @respx.mock
+    def test_no_budget_means_one_request_and_no_refresh(self):
+        route = respx.get(self.URL).mock(return_value=httpx.Response(401))
+        client, provider = self.make_client(max_retries=0)
+
+        with pytest.raises(AuthError):
+            client.get("/thing")
+
+        assert route.call_count == 1
+        assert provider.refreshes == 0
+
+    @respx.mock
+    def test_replay_spends_an_attempt_and_is_actually_sent(self):
+        # The regression the ungated version caused: refresh, then rethrow the
+        # stale 401 WITHOUT ever sending the refreshed request. Two requests
+        # and a 200 body prove the replay reached the wire.
+        route = respx.get(self.URL).mock(side_effect=[httpx.Response(401), httpx.Response(200, json={"ok": True})])
+        client, provider = self.make_client(max_retries=2)
+
+        response = client.get("/thing")
+
+        assert response.status_code == 200
+        assert route.call_count == 2
+        assert provider.refreshes == 1
+
+    @respx.mock
+    def test_replay_and_transient_retries_share_one_budget(self):
+        # 401 (attempt 1, refresh) → 503 (attempt 2) → budget spent. Three
+        # requests would mean the replay rode outside the cap.
+        route = respx.get(self.URL).mock(side_effect=[httpx.Response(401), httpx.Response(503)])
+        client, provider = self.make_client(max_retries=2)
+
+        with pytest.raises(ApiError):
+            client.get("/thing")
+
+        assert route.call_count == 2
+        assert provider.refreshes == 1
+
+    @respx.mock
+    def test_refresh_network_failure_still_retries_under_the_budget(self):
+        """A token endpoint that fails is a transient failure of this attempt.
+
+        Performing the refresh inside the `except AuthError` suite would let
+        this NetworkError bypass the sibling transient handler entirely — an
+        exception raised inside an except suite is not offered to that try's
+        other handlers — ending the request with budget still unspent.
+        """
+        route = respx.get(self.URL).mock(side_effect=[httpx.Response(401), httpx.Response(200, json={"ok": True})])
+        config = Config(base_url="https://3.basecampapi.com", max_retries=3, base_delay=0.001, max_jitter=0.0)
+
+        class ExplodingProvider(StaticTokenProvider):
+            refreshable = True
+
+            def __init__(self):
+                super().__init__("test-token")
+                self.refreshes = 0
+
+            def refresh(self):
+                self.refreshes += 1
+                raise NetworkError("token endpoint timed out")
+
+        provider = ExplodingProvider()
+        client = HttpClient(config, BearerAuth(provider), BasecampHooks())
+
+        response = client.get("/thing")
+
+        assert response.status_code == 200
+        assert route.call_count == 2
+        assert provider.refreshes == 1
+
+    @respx.mock
+    def test_a_throwing_refresh_still_spends_the_one_allowed_refresh(self):
+        """SPEC §4 counts refresh ATTEMPTS, not successes.
+
+        The 401 recurs because the token never changed, so a flag set only on
+        success would let the same request call refresh() a second time — and
+        if the first call reached the server and rotated the token before its
+        response was lost, the second spends a refresh token already dead.
+        """
+        route = respx.get(self.URL).mock(return_value=httpx.Response(401))
+        config = Config(base_url="https://3.basecampapi.com", max_retries=3, base_delay=0.001, max_jitter=0.0)
+
+        class ExplodingProvider(StaticTokenProvider):
+            refreshable = True
+
+            def __init__(self):
+                super().__init__("test-token")
+                self.refreshes = 0
+
+            def refresh(self):
+                self.refreshes += 1
+                raise NetworkError("token endpoint timed out")
+
+        provider = ExplodingProvider()
+        client = HttpClient(config, BearerAuth(provider), BasecampHooks())
+
+        with pytest.raises(AuthError):
+            client.get("/thing")
+
+        # Attempt 1 401s and burns the one refresh; attempt 2 401s with the same
+        # token and must NOT refresh again, so the stale 401 surfaces.
+        assert provider.refreshes == 1
+        assert route.call_count == 2
+
+    @respx.mock
+    def test_mutations_keep_the_in_primitive_replay(self):
+        # Mutations bypass the retry loop entirely (no transient budget), so
+        # they keep the single-request primitive's own replay. SPEC §4's gate
+        # applies where a total-attempt budget governs the path.
+        route = respx.post(self.URL).mock(side_effect=[httpx.Response(401), httpx.Response(201, json={"ok": True})])
+        client, provider = self.make_client(max_retries=0)
+
+        response = client.post("/thing", json_body={"x": 1})
+
+        assert response.status_code == 201
+        assert route.call_count == 2
+        assert provider.refreshes == 1
+
+
+class _RefreshableProvider(StaticTokenProvider):
+    refreshable = True
+
+    def __init__(self):
+        super().__init__("test-token")
+        self.refreshes = 0
+
+    def refresh(self):
+        self.refreshes += 1
+        return True
diff --git a/ruby/lib/basecamp/http.rb b/ruby/lib/basecamp/http.rb
index 1a175f48a..d3b3b2aa4 100644
--- a/ruby/lib/basecamp/http.rb
+++ b/ruby/lib/basecamp/http.rb
@@ -380,26 +380,76 @@ def request_with_retry(method, url, params: {}, allow_cross_origin: false, opera
         @config.max_retries
       end
       attempt = 0
+      refreshed_once = false
       last_error = nil
 
       loop do
         attempt += 1
         break if attempt > max_attempts
 
+        # Each rescue only CLASSIFIES the attempt; the retry side effects run
+        # below, outside every handler. An exception raised inside a rescue
+        # clause bypasses that begin's sibling rescues, so refreshing there
+        # would let a NetworkError from the token endpoint escape the loop
+        # with budget still on the table.
+        error = nil
+        stale_auth = nil
+
         begin
           return single_request(method, url, params: params, body: nil, attempt: attempt,
-            allow_cross_origin: allow_cross_origin, accept: accept)
+            allow_cross_origin: allow_cross_origin, accept: accept, refresh_replay: false)
+        rescue Basecamp::AuthError => e
+          # SPEC §4: the refresh replay is a request on the wire, so it spends
+          # an attempt from THIS budget rather than an uncounted one inside
+          # single_request. max_retries is a total attempt count (#461), and a
+          # cap of one means one request whatever would have caused the second.
+          #
+          # The budget is checked BEFORE refresh so a rotation is never burned
+          # on an attempt the loop has no room to make.
+          raise e if refreshed_once || e.http_status != 401 || attempt >= max_attempts
+
+          stale_auth = e
         rescue Basecamp::RateLimitError, Basecamp::NetworkError, Basecamp::ApiError => e
-          raise e unless retry_eligible?(e, op_retry, retry_on)
+          error = e
+        end
+
+        if stale_auth
+          # SPEC §4 tracks refresh with an "attempted" boolean, not a
+          # "succeeded" one, so mark it BEFORE invoking the provider: a refresh
+          # that raises still counts as the one attempt this request gets.
+          # Otherwise a transient token-endpoint failure lets the NEXT 401 in
+          # the same request call refresh again — and if the first call reached
+          # the server and rotated the token before its response was lost, the
+          # second spends a refresh token that is already dead.
+          refreshed_once = true
+
+          begin
+            refreshed = @token_provider&.refreshable? && @token_provider.refresh
+          rescue Basecamp::RateLimitError, Basecamp::NetworkError, Basecamp::ApiError => e
+            # A token endpoint that times out is a transient failure of this
+            # attempt, not a terminal auth fault: it retries under the same
+            # budget, exactly as it did when the replay lived in single_request.
+            error = e
+          else
+            raise stale_auth unless refreshed
+
+            # No backoff: the token is fresh, and the server never asked us to
+            # wait. The replay still costs the attempt counted above.
+            next
+          end
+        end
+
+        if error
+          raise error unless retry_eligible?(error, op_retry, retry_on)
 
-          last_error = e
+          last_error = error
 
           # Don't sleep if this was the last attempt
           break if attempt >= max_attempts
 
-          delay = calculate_delay(attempt, e.retry_after)
+          delay = calculate_delay(attempt, error.retry_after)
 
-          @hooks.on_retry(RequestInfo.new(method: method.to_s.upcase, url: url, attempt: attempt), attempt + 1, e,
+          @hooks.on_retry(RequestInfo.new(method: method.to_s.upcase, url: url, attempt: attempt), attempt + 1, error,
                           delay)
           sleep(delay)
         end
@@ -434,7 +484,7 @@ def self.operation_retry(operation)
     end
 
     def single_request(method, url, params:, body:, attempt:, retry_count: 0, allow_cross_origin: false,
-      accept: "application/json")
+      accept: "application/json", refresh_replay: true)
       assert_credential_origin!(url, allow_cross_origin)
       info = RequestInfo.new(method: method.to_s.upcase, url: url, attempt: attempt)
       @hooks.on_request_start(info)
@@ -457,7 +507,7 @@ def single_request(method, url, params:, body:, attempt:, retry_count: 0, allow_
         )
       rescue Faraday::ServerError, Faraday::ClientError => e
         duration = Process.clock_gettime(Process::CLOCK_MONOTONIC) - start_time
-        error = handle_error(e)
+        error = handle_error(e, refresh_on_401: refresh_replay)
         result = RequestResult.new(
           status_code: e.response&.dig(:status),
           duration: duration,
@@ -466,11 +516,15 @@ def single_request(method, url, params:, body:, attempt:, retry_count: 0, allow_
         )
         @hooks.on_request_end(info, result)
 
-        # After a successful token refresh on 401, retry the request once
-        if error.is_a?(Basecamp::AuthError) && error.http_status == 401 && retry_count < 1 && @token_refreshed
+        # 401 replay for callers that come here directly (mutations), which
+        # have no retry loop to own it. request_with_retry passes
+        # refresh_replay: false and replays from the loop so the extra request
+        # draws from the attempt budget (SPEC §4).
+        if refresh_replay && error.is_a?(Basecamp::AuthError) && error.http_status == 401 && retry_count < 1 \
+            && @token_refreshed
           @token_refreshed = false
           return single_request(method, url, params: params, body: body, attempt: attempt, retry_count: retry_count + 1,
-            allow_cross_origin: allow_cross_origin, accept: accept)
+            allow_cross_origin: allow_cross_origin, accept: accept, refresh_replay: refresh_replay)
         end
 
         raise error
@@ -533,7 +587,10 @@ def single_request_raw(method, url, body:, content_type:, attempt:)
       end
     end
 
-    def handle_error(error)
+    # refresh_on_401: false leaves the token alone — the caller (request_with_retry)
+    # owns the refresh so it can gate it on the attempt budget before spending a
+    # rotation. Classifying an error should not rotate credentials as a side effect.
+    def handle_error(error, refresh_on_401: true)
       status = error.response&.dig(:status)
       body = error.response&.dig(:body)
       headers = error.response&.dig(:headers) || {}
@@ -544,7 +601,7 @@ def handle_error(error)
       err = case status
       when 401
         # Try token refresh; flag for caller to retry
-        @token_refreshed = @token_provider&.refreshable? && @token_provider.refresh
+        @token_refreshed = refresh_on_401 && @token_provider&.refreshable? && @token_provider.refresh
         Basecamp::AuthError.new("Authentication failed")
       when 403
         Basecamp::ForbiddenError.new("Access denied")
diff --git a/ruby/test/basecamp/download_test.rb b/ruby/test/basecamp/download_test.rb
index 93648d493..893358444 100644
--- a/ruby/test/basecamp/download_test.rb
+++ b/ruby/test/basecamp/download_test.rb
@@ -438,13 +438,8 @@ def test_download_url_auth_on_every_hop1_attempt_never_on_hop2
     assert_requested(s3_stub)
   end
 
-  # SPEC §4: refresh is attempted at most once PER REQUEST — tracked with a
-  # boolean, not a counter, and not subordinate to the transient-retry budget.
-  # So a refreshable 401 replays once even with retries disabled, deliberately
-  # orthogonal to §14's hop-1 attempt cap. Whether the two should be
-  # reconciled is tracked in #565; this pins what the spec says today.
-  def test_download_url_refreshable_401_replays_once_independent_of_the_retry_cap
-    provider = Class.new do
+  def refreshable_provider
+    Class.new do
       attr_reader :refreshes
 
       def initialize = @refreshes = 0
@@ -456,14 +451,56 @@ def refresh
         true
       end
     end.new
+  end
 
+  # SPEC §4/§14: with retry disabled ONE request goes out — refresh included.
+  # The replay is a request on the wire, so it spends an attempt from the same
+  # total-attempt budget as a transient retry (#565, #461).
+  #
+  # The refresh itself must not fire either: §4 checks the budget BEFORE
+  # refreshing, because rotating a token the SDK has no attempt left to use
+  # burns it for nothing and still surfaces the stale 401.
+  def test_download_url_refresh_replay_is_not_attempted_without_budget
+    provider = refreshable_provider
     stub_request(:get, "#{base_url}#{HOP1_PATH}").to_return(status: 401)
 
     account = create_account_client(config: fast_download_config(max_retries: 0), token_provider: provider)
-    assert_raises(Basecamp::Error) { account.download_url(HOP1_URL) }
+    assert_raises(Basecamp::AuthError) { account.download_url(HOP1_URL) }
+
+    assert_requested(:get, "#{base_url}#{HOP1_PATH}", times: 1)
+    assert_equal 0, provider.refreshes
+  end
+
+  # A budget of two allows the replay, and it consumes attempt 2.
+  def test_download_url_refresh_replay_spends_an_attempt_from_the_budget
+    provider = refreshable_provider
+    stub_request(:get, "#{base_url}#{HOP1_PATH}").to_return(status: 401)
+
+    account = create_account_client(config: fast_download_config(max_retries: 2), token_provider: provider)
+    assert_raises(Basecamp::AuthError) { account.download_url(HOP1_URL) }
 
-    # One transient attempt, plus §4's single refresh replay.
+    # Two requests, not three: the replay drew from the same budget, and §4
+    # allows only one refresh per request regardless.
     assert_requested(:get, "#{base_url}#{HOP1_PATH}", times: 2)
+    assert_equal 1, provider.refreshes
+  end
+
+  # The refreshed request is actually SENT, not refreshed-then-discarded.
+  def test_download_url_refresh_replay_succeeds_when_the_new_token_works
+    provider = refreshable_provider
+    stub_request(:get, "#{base_url}#{HOP1_PATH}")
+      .to_return(status: 401)
+      .then.to_return(status: 302, headers: { "Location" => SIGNED_URL })
+    s3_stub = stub_request(:get, SIGNED_URL)
+      .to_return(status: 200, body: "data", headers: { "Content-Type" => "application/octet-stream" })
+
+    account = create_account_client(config: fast_download_config(max_retries: 2), token_provider: provider)
+    result = account.download_url(HOP1_URL)
+
+    assert_equal "data", result.body
+    assert_requested(:get, "#{base_url}#{HOP1_PATH}", times: 2)
+    assert_requested(s3_stub)
+    assert_equal 1, provider.refreshes
   end
 
   # SPEC §14: hop 1 carries Authorization and User-Agent only. A binary
diff --git a/ruby/test/basecamp/http_test.rb b/ruby/test/basecamp/http_test.rb
index 7fd6b238e..fa5602464 100644
--- a/ruby/test/basecamp/http_test.rb
+++ b/ruby/test/basecamp/http_test.rb
@@ -154,6 +154,127 @@ def test_401_refresh_and_retry_no_infinite_loop
     assert_requested(:get, "https://3.basecampapi.com/test.json", times: 2)
   end
 
+  # SPEC §4: the refresh replay is a request on the wire, so it spends an
+  # attempt from the same total-attempt budget as a transient retry (#565,
+  # #461). These pin the PLAIN GET path — the "every GET in Python and Ruby"
+  # case that an earlier, ungated attempt at this change regressed.
+  def test_401_refresh_replay_is_not_attempted_without_budget
+    provider = RefreshableTokenProvider.new("old-token", refresh_result: true)
+    config = Basecamp::Config.new(base_url: "https://3.basecampapi.com", timeout: 5, max_retries: 1)
+    http = Basecamp::Http.new(config: config, token_provider: provider)
+
+    stub_request(:get, "https://3.basecampapi.com/test.json")
+      .to_return(status: 401, body: '{"error": "Unauthorized"}')
+
+    assert_raises(Basecamp::AuthError) { http.get("/test.json") }
+
+    # One request, and no rotation burned on an attempt the loop cannot make.
+    assert_requested(:get, "https://3.basecampapi.com/test.json", times: 1)
+    assert_equal 0, provider.refresh_count
+  end
+
+  def test_401_refresh_replay_shares_the_budget_with_transient_retries
+    provider = RefreshableTokenProvider.new("old-token", refresh_result: true)
+    config = Basecamp::Config.new(base_url: "https://3.basecampapi.com", timeout: 5, max_retries: 2,
+      base_delay: 0.001, max_jitter: 0.0)
+    http = Basecamp::Http.new(config: config, token_provider: provider)
+
+    stub_request(:get, "https://3.basecampapi.com/test.json")
+      .to_return(status: 401, body: '{"error": "Unauthorized"}')
+      .then.to_return(status: 503, body: '{"error": "Unavailable"}')
+
+    assert_raises(Basecamp::ApiError) { http.get("/test.json") }
+
+    # 401 (attempt 1, refresh) then 503 (attempt 2) exhausts the cap. Three
+    # requests would mean the replay rode outside it.
+    assert_requested(:get, "https://3.basecampapi.com/test.json", times: 2)
+    assert_equal 1, provider.refresh_count
+  end
+
+  def test_401_refresh_replay_is_actually_sent_not_discarded
+    # The regression an ungated version caused: refresh, then rethrow the stale
+    # 401 without ever sending the refreshed request. A 200 body proves it went.
+    provider = RefreshableTokenProvider.new("old-token", refresh_result: true)
+    config = Basecamp::Config.new(base_url: "https://3.basecampapi.com", timeout: 5, max_retries: 2)
+    http = Basecamp::Http.new(config: config, token_provider: provider)
+
+    stub_request(:get, "https://3.basecampapi.com/test.json")
+      .to_return(status: 401, body: '{"error": "Unauthorized"}')
+      .then.to_return(status: 200, body: '{"ok": true}')
+
+    response = http.get("/test.json")
+
+    assert_equal 200, response.status
+    assert_requested(:get, "https://3.basecampapi.com/test.json", times: 2)
+    assert_equal 1, provider.refresh_count
+  end
+
+  # A token endpoint that fails is a transient failure of THIS attempt.
+  # Performing the refresh inside the `rescue Basecamp::AuthError` clause would
+  # let this NetworkError bypass the sibling transient rescue entirely — an
+  # exception raised inside a rescue is not offered to that begin's other
+  # rescues — ending the request with budget still unspent.
+  def test_401_refresh_network_failure_still_retries_under_the_budget
+    provider = Class.new do
+      attr_reader :refresh_count
+
+      def initialize = @refresh_count = 0
+      def access_token = "old-token"
+      def refreshable? = true
+
+      def refresh
+        @refresh_count += 1
+        raise Basecamp::NetworkError.new("token endpoint timed out")
+      end
+    end.new
+
+    config = Basecamp::Config.new(base_url: "https://3.basecampapi.com", timeout: 5, max_retries: 3,
+      base_delay: 0.001, max_jitter: 0.0)
+    http = Basecamp::Http.new(config: config, token_provider: provider)
+
+    stub_request(:get, "https://3.basecampapi.com/test.json")
+      .to_return(status: 401, body: '{"error": "Unauthorized"}')
+      .then.to_return(status: 200, body: '{"ok": true}')
+
+    response = http.get("/test.json")
+
+    assert_equal 200, response.status
+    assert_requested(:get, "https://3.basecampapi.com/test.json", times: 2)
+    assert_equal 1, provider.refresh_count
+  end
+
+  # SPEC §4 counts refresh ATTEMPTS, not successes. The 401 recurs because the
+  # token never changed, so a flag set only on success would let the same
+  # request call refresh a second time — and if the first call reached the
+  # server and rotated the token before its response was lost, the second
+  # spends a refresh token that is already dead.
+  def test_401_throwing_refresh_still_spends_the_one_allowed_refresh
+    provider = Class.new do
+      attr_reader :refresh_count
+
+      def initialize = @refresh_count = 0
+      def access_token = "old-token"
+      def refreshable? = true
+
+      def refresh
+        @refresh_count += 1
+        raise Basecamp::NetworkError.new("token endpoint timed out")
+      end
+    end.new
+
+    config = Basecamp::Config.new(base_url: "https://3.basecampapi.com", timeout: 5, max_retries: 3,
+      base_delay: 0.001, max_jitter: 0.0)
+    http = Basecamp::Http.new(config: config, token_provider: provider)
+
+    stub_request(:get, "https://3.basecampapi.com/test.json")
+      .to_return(status: 401, body: '{"error": "Unauthorized"}')
+
+    assert_raises(Basecamp::AuthError) { http.get("/test.json") }
+
+    assert_equal 1, provider.refresh_count
+    assert_requested(:get, "https://3.basecampapi.com/test.json", times: 2)
+  end
+
   def test_401_no_retry_when_refresh_fails
     provider = RefreshableTokenProvider.new("old-token", refresh_result: false)
     http = Basecamp::Http.new(config: @config, token_provider: provider)