From c55b6641646a9959b0a53b1ad179a4ee7fe3201d Mon Sep 17 00:00:00 2001
From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com>
Date: Fri, 14 Aug 2026 16:46:00 +0300
Subject: [PATCH 01/14] feat(low-code): reconcile
 RateLimitedMultipleTokenAuthenticator quota against response headers

The authenticator seeds each token's quota counters from `quota_status_url` and then
decrements them locally, never reconciling against what the server reports. When the
local view and the server disagree the component stops doing the one thing it exists
for -- spreading load across tokens -- because rotation only triggers on its own
bookkeeping reaching zero.

Measured on source-github with two tokens both seeded at 5000 remaining, against an
endpoint returning 403 with `X-RateLimit-Remaining: 0` and a reset 30 minutes out:
the declarative path retried the same token three times and slept 2 x 1799.6s while
the second, fully-quota'd token sat idle. The equivalent Python stream, which rotates
from its backoff strategy, finished in seconds.

`TokenQuota` gains four optional fields -- `remaining_header`, `reset_header`,
`limit_header` and `exhaustion_status_codes` -- and `HttpClient` calls
`update_from_response` on the authenticator after every attempt, mirroring what
`LimiterMixin.send` already does for `api_budget`. Reactive rotation then falls out of
the existing proactive path: the server says the pool is spent, the counter goes to
zero, and the next request picks another token. No new coupling between error handlers
and authenticators, and the underlying counter drift is fixed rather than worked
around.

Update rules, which differ from `HttpAPIBudget`'s straight assignment because these
counters are per token and mutated concurrently:

- Response reset later than the stored one: the window rolled over, so take the
  server's numbers wholesale.
- Same window: `min(stored, header)`. Responses arrive out of order and a slow one
  carries a stale, higher count; handing those calls back would let concurrent
  requests overspend.
- Response reset earlier than stored: ignore, it belongs to a previous window.
- Status in `exhaustion_status_codes` with no remaining header: zero the pool.

Updates are attributed to the token that signed the request rather than the currently
active one, since the authenticator may rotate between a request going out and its
response arriving. Cached responses are skipped -- a replay carries stale headers and
consumed no quota. A failure inside the update is logged at debug and swallowed:
quota bookkeeping must never turn a good response into a sync failure.

All four fields are optional and default to unset, so pools configured only with
paths behave exactly as before.

Out of scope: secondary rate limits, which on GitHub do not move
`X-RateLimit-Remaining` and arrive as 403/429 with `Retry-After`. Listing 403 in
`exhaustion_status_codes` would park a healthy token on a plain permission error, so a
dedicated cooldown field is the better follow-up.

Note on `models/declarative_component_schema.py`: regenerating produces ~280 lines of
unrelated churn because the committed file is already stale relative to the current
generator (verified by regenerating from an unmodified main). Only the `TokenQuota`
class is updated here, byte-identical to what the generator emits for it, so the diff
stays attributable. The pre-existing drift is worth a separate cleanup.

Testing: 16 new authenticator tests and 3 new `HttpClient` tests; 1601 tests pass
across auth, parsers, requesters and streams. Two failures in
`test_concurrent_declarative_source.py` reproduce identically on unmodified main
(requests_cache SQLite locking) and are unrelated.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../auth/rate_limited_multiple_token.py       | 103 ++++++-
 .../declarative_component_schema.yaml         |  26 ++
 .../models/declarative_component_schema.py    |  24 ++
 .../parsers/model_to_component_factory.py     |   8 +
 .../sources/streams/http/http_client.py       |  28 ++
 .../auth/test_rate_limited_multiple_token.py  | 282 ++++++++++++++++++
 .../sources/streams/http/test_http_client.py  |  69 +++++
 7 files changed, 538 insertions(+), 2 deletions(-)

diff --git a/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py b/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py
index a07e6a36d8..ff95bce176 100644
--- a/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py
+++ b/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py
@@ -28,6 +28,12 @@ class TokenQuota:
     `matchers` classify outgoing requests into the pool; a pool with no matchers acts as the
     default pool. `remaining_path`/`reset_path`/`limit_path` locate the pool's values in the
     quota status response.
+
+    `remaining_header`/`reset_header`/`limit_header`/`exhaustion_status_codes` are optional and
+    enable reconciling the pool against what the server reports on each response. Without them
+    the pool is only ever seeded from `quota_status_url`, so its counters drift whenever the
+    token is shared with another client, requests are in flight concurrently, or the sync runs
+    long enough for a single seeding to go stale.
     """
 
     name: str
@@ -35,6 +41,14 @@ class TokenQuota:
     reset_path: List[str]
     limit_path: Optional[List[str]] = None
     matchers: List[RequestMatcher] = field(default_factory=list)
+    remaining_header: Optional[str] = None
+    reset_header: Optional[str] = None
+    limit_header: Optional[str] = None
+    exhaustion_status_codes: List[int] = field(default_factory=list)
+
+    @property
+    def is_response_aware(self) -> bool:
+        return bool(self.remaining_header or self.limit_header or self.exhaustion_status_codes)
 
 
 @dataclass
@@ -59,8 +73,11 @@ class RateLimitedMultipleTokenAuthenticator(DeclarativeAuthenticator):
     `seconds_until_reset / total_remaining` (capped at 10s) is injected before each request.
 
     Counters are seeded per token from `quota_status_url` on first use and refreshed after an
-    exhaustion wait. All state transitions are guarded by a lock so the authenticator can be
-    shared safely across concurrent streams; sleeps never hold the lock.
+    exhaustion wait. When a pool declares response headers, `update_from_response` additionally
+    reconciles that pool against the server on every response, which keeps the counters honest
+    between seedings and makes the authenticator rotate off a token the server has rejected even
+    though the local count still looks healthy. All state transitions are guarded by a lock so
+    the authenticator can be shared safely across concurrent streams; sleeps never hold the lock.
     """
 
     HEARTBEAT_INTERVAL = 60.0  # Log every 60s during exhaustion wait
@@ -319,3 +336,85 @@ def _extract_path(self, response_body: Mapping[str, Any], path: List[str]) -> An
                 )
             value = value[key]
         return value
+
+    def update_from_response(
+        self, request: requests.PreparedRequest, response: requests.Response
+    ) -> None:
+        """Reconcile the matched pool's counters against what the server reported.
+
+        Called once per HTTP attempt by `HttpClient`. Inert unless the matched pool declares
+        response headers, so behaviour is unchanged for pools configured only with paths.
+
+        The update is attributed to the token that *sent* the request rather than to the
+        currently active one: under concurrency the authenticator may have rotated between the
+        request going out and the response coming back.
+        """
+        quota = self._match_quota(request)
+        if not quota.is_response_aware:
+            return
+        token = self._token_from_request(request)
+        if token is None:
+            return
+
+        remaining = self._header_int(response, quota.remaining_header)
+        if remaining is None and response.status_code in quota.exhaustion_status_codes:
+            # The server rejected the call for rate-limit reasons but told us nothing about the
+            # remaining count. Treat the pool as spent so the next request rotates.
+            remaining = 0
+        reset_at = self._header_datetime(response, quota.reset_header)
+        limit = self._header_int(response, quota.limit_header)
+        if remaining is None and limit is None:
+            return
+
+        with self._lock:
+            state = self._states.get(token, {}).get(quota.name)
+            if state is None:
+                return  # not seeded yet; the initial seeding is the more authoritative source
+            if limit is not None and limit > 0:
+                state.limit = limit
+            if reset_at is not None and reset_at > state.reset_at:
+                # The quota window rolled over, so the local count is meaningless -- take the
+                # server's numbers wholesale.
+                state.reset_at = reset_at
+                if remaining is not None:
+                    state.remaining = remaining
+            elif remaining is not None and (reset_at is None or reset_at == state.reset_at):
+                # Same window. Responses can arrive out of order and a slow one carries a stale,
+                # higher count, so only ever tighten the estimate -- never hand back calls that
+                # concurrent requests have already spent.
+                state.remaining = min(state.remaining, remaining)
+
+    def _token_from_request(self, request: requests.PreparedRequest) -> Optional[str]:
+        """Recover the token a request was signed with from its auth header."""
+        value = request.headers.get(self._header)
+        if not value:
+            return None
+        token = value[len(self._auth_method) :].strip() if self._auth_method else value.strip()
+        return token if token in self._states else None
+
+    @staticmethod
+    def _header_int(response: requests.Response, header: Optional[str]) -> Optional[int]:
+        if not header:
+            return None
+        value = response.headers.get(header)
+        if value is None:
+            return None
+        try:
+            return int(value)
+        except (TypeError, ValueError):
+            return None
+
+    @staticmethod
+    def _header_datetime(
+        response: requests.Response, header: Optional[str]
+    ) -> Optional[AirbyteDateTime]:
+        if not header:
+            return None
+        value = response.headers.get(header)
+        if value is None:
+            return None
+        try:
+            # Same parsing rules as `reset_path`, so epoch seconds and ISO 8601 both work.
+            return ab_datetime_parse(value)
+        except Exception:
+            return None
diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml
index 97dda3b98b..06bfed587c 100644
--- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml
+++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml
@@ -497,6 +497,32 @@ definitions:
         type: array
         items:
           "$ref": "#/definitions/HttpRequestRegexMatcher"
+      remaining_header:
+        title: Remaining Header
+        description: Optional response header carrying the remaining call count for this pool. When set, the pool's counter is reconciled against this header on every response, which corrects drift caused by sharing the token with other clients, by requests in flight concurrently, or by a sync running long enough for the initial quota status read to go stale. Without it the pool is only ever seeded from the quota status endpoint.
+        type: string
+        examples:
+          - "X-RateLimit-Remaining"
+      reset_header:
+        title: Reset Header
+        description: Optional response header carrying the quota reset timestamp for this pool. Parsed with the same rules as `reset_path`, so epoch seconds and ISO 8601 both work. Used to tell a rolled-over quota window from the current one.
+        type: string
+        examples:
+          - "X-RateLimit-Reset"
+      limit_header:
+        title: Limit Header
+        description: Optional response header carrying the total call limit for this pool, used to keep the proactive throttling reserve accurate as the limit changes.
+        type: string
+        examples:
+          - "X-RateLimit-Limit"
+      exhaustion_status_codes:
+        title: Exhaustion Status Codes
+        description: Response status codes that mean this token's pool is spent, applied when the response carries no remaining header. The pool is set to zero so the next request rotates to another token instead of waiting out the reset window. Only list codes the API uses exclusively for rate limiting -- a code that also signals other failures would park a healthy token.
+        type: array
+        items:
+          type: integer
+        examples:
+          - [429]
       $parameters:
         type: object
         additionalProperties: true
diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py
index a84412eaf9..f1b4c225b3 100644
--- a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py
+++ b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py
@@ -609,6 +609,30 @@ class TokenQuota(BaseModel):
         description="List of matchers that classify outgoing requests into this quota pool. The first pool whose matcher matches a request is used. A pool with no matchers acts as the default pool.",
         title="Matchers",
     )
+    remaining_header: Optional[str] = Field(
+        None,
+        description="Optional response header carrying the remaining call count for this pool. When set, the pool's counter is reconciled against this header on every response, which corrects drift caused by sharing the token with other clients, by requests in flight concurrently, or by a sync running long enough for the initial quota status read to go stale. Without it the pool is only ever seeded from the quota status endpoint.",
+        examples=["X-RateLimit-Remaining"],
+        title="Remaining Header",
+    )
+    reset_header: Optional[str] = Field(
+        None,
+        description="Optional response header carrying the quota reset timestamp for this pool. Parsed with the same rules as `reset_path`, so epoch seconds and ISO 8601 both work. Used to tell a rolled-over quota window from the current one.",
+        examples=["X-RateLimit-Reset"],
+        title="Reset Header",
+    )
+    limit_header: Optional[str] = Field(
+        None,
+        description="Optional response header carrying the total call limit for this pool, used to keep the proactive throttling reserve accurate as the limit changes.",
+        examples=["X-RateLimit-Limit"],
+        title="Limit Header",
+    )
+    exhaustion_status_codes: Optional[List[int]] = Field(
+        None,
+        description="Response status codes that mean this token's pool is spent, applied when the response carries no remaining header. The pool is set to zero so the next request rotates to another token instead of waiting out the reset window. Only list codes the API uses exclusively for rate limiting -- a code that also signals other failures would park a healthy token.",
+        examples=[[429]],
+        title="Exhaustion Status Codes",
+    )
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 
diff --git a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py
index b992b4201a..d9d3ecfc8f 100644
--- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py
+++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py
@@ -4660,6 +4660,10 @@ def create_rate_limited_multiple_token_authenticator(
                 "remaining_path": quota_model.remaining_path,
                 "reset_path": quota_model.reset_path,
                 "limit_path": quota_model.limit_path,
+                "remaining_header": quota_model.remaining_header,
+                "reset_header": quota_model.reset_header,
+                "limit_header": quota_model.limit_header,
+                "exhaustion_status_codes": quota_model.exhaustion_status_codes,
                 "matchers": [
                     {
                         "method": matcher_model.method,
@@ -4734,6 +4738,10 @@ def create_rate_limited_multiple_token_authenticator(
                 remaining_path=quota_model.remaining_path,
                 reset_path=quota_model.reset_path,
                 limit_path=quota_model.limit_path,
+                remaining_header=quota_model.remaining_header,
+                reset_header=quota_model.reset_header,
+                limit_header=quota_model.limit_header,
+                exhaustion_status_codes=quota_model.exhaustion_status_codes or [],
                 matchers=[
                     self.create_http_request_matcher(matcher_model, config)
                     for matcher_model in quota_model.matchers or []
diff --git a/airbyte_cdk/sources/streams/http/http_client.py b/airbyte_cdk/sources/streams/http/http_client.py
index c1d0eabd67..9b21a12af4 100644
--- a/airbyte_cdk/sources/streams/http/http_client.py
+++ b/airbyte_cdk/sources/streams/http/http_client.py
@@ -322,6 +322,31 @@ def _send_with_retry(
                 stream_descriptor=StreamDescriptor(name=self._name),
             )
 
+    def _update_authenticator_from_response(
+        self, request: requests.PreparedRequest, response: requests.Response
+    ) -> None:
+        """Let a quota-tracking authenticator reconcile its state against the server.
+
+        Authenticators only ever see requests, so an authenticator that tracks per-token quota
+        has no way to learn that the server disagrees with its local bookkeeping. This is the
+        feedback channel, mirroring what `LimiterMixin.send` does for the API budget.
+        """
+        authenticator = getattr(self._session, "auth", None)
+        update_from_response = getattr(authenticator, "update_from_response", None)
+        if not callable(update_from_response):
+            return
+        if getattr(response, "from_cache", False):
+            # A replayed cached response carries the rate-limit headers from whenever it was
+            # first fetched and consumed no quota of its own.
+            return
+        try:
+            update_from_response(request, response)
+        except Exception:
+            # Quota bookkeeping must never turn an otherwise fine response into a failure.
+            self._logger.debug(
+                "Authenticator failed to update quota state from response", exc_info=True
+            )
+
     def _send(
         self,
         request: requests.PreparedRequest,
@@ -349,6 +374,9 @@ def _send(
         except requests.RequestException as e:
             exc = e
 
+        if response is not None:
+            self._update_authenticator_from_response(request, response)
+
         error_resolution: ErrorResolution = self._error_handler.interpret_response(
             response if response is not None else exc
         )
diff --git a/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py b/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py
index 395b9b1727..5220d534be 100644
--- a/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py
+++ b/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py
@@ -281,6 +281,249 @@ def fake_sleep(seconds):
     assert clock["now"] <= 30
 
 
+# --- reconciling quota state against response headers ---------------------------------------
+
+
+def _response_aware_quotas(exhaustion_status_codes=(429,)):
+    return [
+        TokenQuota(
+            name="rest",
+            remaining_path=["resources", "core", "remaining"],
+            reset_path=["resources", "core", "reset"],
+            limit_path=["resources", "core", "limit"],
+            remaining_header="X-RateLimit-Remaining",
+            reset_header="X-RateLimit-Reset",
+            limit_header="X-RateLimit-Limit",
+            exhaustion_status_codes=list(exhaustion_status_codes),
+        ),
+        TokenQuota(
+            name="graphql",
+            remaining_path=["resources", "graphql", "remaining"],
+            reset_path=["resources", "graphql", "reset"],
+            limit_path=["resources", "graphql", "limit"],
+            remaining_header="X-RateLimit-Remaining",
+            reset_header="X-RateLimit-Reset",
+            matchers=[HttpRequestRegexMatcher(url_path_pattern="/graphql")],
+        ),
+    ]
+
+
+def _response_aware_authenticator(tokens=("token_1", "token_2"), **kwargs):
+    return RateLimitedMultipleTokenAuthenticator(
+        tokens=list(tokens),
+        quotas=_response_aware_quotas(**kwargs),
+        quota_status_url=QUOTA_STATUS_URL,
+        auth_method="token",
+    )
+
+
+def _response(status_code=200, headers=None, from_cache=False):
+    response = requests.Response()
+    response.status_code = status_code
+    response.headers.update(headers or {})
+    if from_cache:
+        response.from_cache = True  # type: ignore[attr-defined] # mirrors requests_cache
+    return response
+
+
+def test_response_header_corrects_counter_downward(requests_mock):
+    """The server is the source of truth: a lower remaining count wins over the local estimate."""
+    requests_mock.get(QUOTA_STATUS_URL, json=_quota_status_body())
+    authenticator = _response_aware_authenticator(tokens=("token_1",))
+    request = authenticator(_prepared_request())
+    assert authenticator._states["token_1"]["rest"].remaining == 4999
+
+    authenticator.update_from_response(request, _response(headers={"X-RateLimit-Remaining": "10"}))
+
+    assert authenticator._states["token_1"]["rest"].remaining == 10
+
+
+def test_response_header_does_not_ratchet_counter_up_within_a_window(requests_mock):
+    """A slow response carries a stale, higher count. Handing those calls back would let concurrent
+    requests overspend, so within one window the estimate may only tighten."""
+    requests_mock.get(QUOTA_STATUS_URL, json=_quota_status_body())
+    authenticator = _response_aware_authenticator(tokens=("token_1",))
+    request = authenticator(_prepared_request())
+    reset_at = authenticator._states["token_1"]["rest"].reset_at
+    authenticator._states["token_1"]["rest"].remaining = 10
+
+    authenticator.update_from_response(
+        request,
+        _response(
+            headers={
+                "X-RateLimit-Remaining": "4000",
+                "X-RateLimit-Reset": str(int(reset_at.timestamp())),
+            }
+        ),
+    )
+
+    assert authenticator._states["token_1"]["rest"].remaining == 10
+
+
+def test_later_reset_in_response_starts_a_new_window(requests_mock):
+    requests_mock.get(QUOTA_STATUS_URL, json=_quota_status_body())
+    authenticator = _response_aware_authenticator(tokens=("token_1",))
+    request = authenticator(_prepared_request())
+    state = authenticator._states["token_1"]["rest"]
+    state.remaining = 3
+    next_window = int(state.reset_at.timestamp()) + 3600
+
+    authenticator.update_from_response(
+        request,
+        _response(headers={"X-RateLimit-Remaining": "5000", "X-RateLimit-Reset": str(next_window)}),
+    )
+
+    assert state.remaining == 5000
+    assert int(state.reset_at.timestamp()) == next_window
+
+
+def test_earlier_reset_in_response_is_ignored(requests_mock):
+    """A response from a window that has already rolled over must not disturb current state."""
+    requests_mock.get(QUOTA_STATUS_URL, json=_quota_status_body())
+    authenticator = _response_aware_authenticator(tokens=("token_1",))
+    request = authenticator(_prepared_request())
+    state = authenticator._states["token_1"]["rest"]
+    state.remaining = 100
+    previous_window = int(state.reset_at.timestamp()) - 3600
+
+    authenticator.update_from_response(
+        request,
+        _response(
+            headers={"X-RateLimit-Remaining": "7", "X-RateLimit-Reset": str(previous_window)}
+        ),
+    )
+
+    assert state.remaining == 100
+
+
+def test_limit_header_updates_the_throttling_reserve(requests_mock):
+    requests_mock.get(QUOTA_STATUS_URL, json=_quota_status_body())
+    authenticator = _response_aware_authenticator(tokens=("token_1",))
+    request = authenticator(_prepared_request())
+
+    authenticator.update_from_response(request, _response(headers={"X-RateLimit-Limit": "15000"}))
+
+    assert authenticator._states["token_1"]["rest"].limit == 15000
+
+
+def test_exhaustion_status_code_zeroes_the_pool_and_next_request_rotates(requests_mock):
+    """Regression test for the core defect.
+
+    GitHub rejects a request for rate-limit reasons while the local counter still reads healthy.
+    Before response awareness the authenticator kept handing out the same token and the error
+    handler waited out the whole reset window with other tokens sitting idle.
+    """
+    requests_mock.get(QUOTA_STATUS_URL, json=_quota_status_body())
+    authenticator = _response_aware_authenticator()
+    request = authenticator(_prepared_request())
+    assert request.headers["Authorization"] == "token token_1"
+    assert authenticator._states["token_1"]["rest"].remaining == 4999  # locally still healthy
+
+    authenticator.update_from_response(request, _response(status_code=429))
+
+    assert authenticator._states["token_1"]["rest"].remaining == 0
+    retry = authenticator(_prepared_request())
+    assert retry.headers["Authorization"] == "token token_2"
+
+
+def test_rate_limited_response_is_ignored_when_no_headers_are_configured(requests_mock):
+    """Pools configured only with paths keep their previous behaviour."""
+    requests_mock.get(QUOTA_STATUS_URL, json=_quota_status_body())
+    authenticator = _authenticator(tokens=("token_1", "token_2"))
+    request = authenticator(_prepared_request())
+
+    authenticator.update_from_response(
+        request, _response(status_code=429, headers={"X-RateLimit-Remaining": "0"})
+    )
+
+    assert authenticator._states["token_1"]["rest"].remaining == 4999
+    assert authenticator(_prepared_request()).headers["Authorization"] == "token token_1"
+
+
+def test_response_is_attributed_to_the_token_that_sent_it(requests_mock):
+    """Under concurrency the active token can move on before a response lands, so the update has
+    to follow the request's own auth header rather than whichever token is active now."""
+    requests_mock.get(QUOTA_STATUS_URL, json=_quota_status_body())
+    authenticator = _response_aware_authenticator()
+    request = authenticator(_prepared_request())
+    assert request.headers["Authorization"] == "token token_1"
+    authenticator.update_token = None  # guard against accidental reliance on a rotation helper
+    authenticator._active_token = "token_2"
+
+    authenticator.update_from_response(request, _response(headers={"X-RateLimit-Remaining": "12"}))
+
+    assert authenticator._states["token_1"]["rest"].remaining == 12
+    assert authenticator._states["token_2"]["rest"].remaining == 5000
+
+
+def test_cached_response_does_not_update_quota_state(requests_mock):
+    """A replayed cached response carries stale headers and consumed no quota. `HttpClient` filters
+    those out; this pins the behaviour if anything ever calls the method directly."""
+    requests_mock.get(QUOTA_STATUS_URL, json=_quota_status_body())
+    authenticator = _response_aware_authenticator(tokens=("token_1",))
+    request = authenticator(_prepared_request())
+
+    from airbyte_cdk.sources.streams.http import HttpClient
+
+    client = HttpClient(name="test", logger=__import__("logging").getLogger("test"))
+    client._session.auth = authenticator
+    client._update_authenticator_from_response(
+        request, _response(headers={"X-RateLimit-Remaining": "1"}, from_cache=True)
+    )
+
+    assert authenticator._states["token_1"]["rest"].remaining == 4999
+
+
+def test_graphql_response_updates_only_the_graphql_pool(requests_mock):
+    requests_mock.get(QUOTA_STATUS_URL, json=_quota_status_body())
+    authenticator = _response_aware_authenticator(tokens=("token_1",))
+    request = authenticator(_prepared_request("https://api.example.com/graphql"))
+
+    authenticator.update_from_response(request, _response(headers={"X-RateLimit-Remaining": "42"}))
+
+    assert authenticator._states["token_1"]["graphql"].remaining == 42
+    assert authenticator._states["token_1"]["rest"].remaining == 5000
+
+
+def test_malformed_headers_are_ignored(requests_mock):
+    requests_mock.get(QUOTA_STATUS_URL, json=_quota_status_body())
+    authenticator = _response_aware_authenticator(tokens=("token_1",))
+    request = authenticator(_prepared_request())
+
+    authenticator.update_from_response(
+        request,
+        _response(headers={"X-RateLimit-Remaining": "not-a-number", "X-RateLimit-Reset": "soon"}),
+    )
+
+    assert authenticator._states["token_1"]["rest"].remaining == 4999
+
+
+def test_concurrent_response_updates_do_not_lose_decrements(requests_mock):
+    requests_mock.get(QUOTA_STATUS_URL, json=_quota_status_body())
+    authenticator = _response_aware_authenticator(tokens=("token_1",))
+    authenticator._ensure_initialized()
+    reset_at = authenticator._states["token_1"]["rest"].reset_at
+    headers = {
+        "X-RateLimit-Remaining": "4000",
+        "X-RateLimit-Reset": str(int(reset_at.timestamp())),
+    }
+
+    def worker():
+        for _ in range(50):
+            request = authenticator(_prepared_request())
+            authenticator.update_from_response(request, _response(headers=headers))
+
+    threads = [threading.Thread(target=worker) for _ in range(4)]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join()
+
+    # 200 requests were charged; the server's 4000 clamps the count and further local decrements
+    # continue from there, so the result must never exceed the clamp.
+    assert authenticator._states["token_1"]["rest"].remaining <= 4000
+
+
 def _model(tokens):
     return RateLimitedMultipleTokenAuthenticatorModel.parse_obj(
         {
@@ -422,3 +665,42 @@ def test_factory_interpolates_max_wait_time():
     )
 
     assert authenticator._max_wait_time == timedelta(minutes=30)
+
+
+def test_factory_passes_response_header_fields_through():
+    factory = ModelToComponentFactory()
+    model = _model("{{ config['pat'] }}")
+    model.quotas[0].remaining_header = "X-RateLimit-Remaining"
+    model.quotas[0].reset_header = "X-RateLimit-Reset"
+    model.quotas[0].limit_header = "X-RateLimit-Limit"
+    model.quotas[0].exhaustion_status_codes = [429]
+
+    authenticator = factory.create_rate_limited_multiple_token_authenticator(
+        model, {"pat": "token_1"}
+    )
+
+    rest, graphql = authenticator._quotas
+    assert rest.remaining_header == "X-RateLimit-Remaining"
+    assert rest.reset_header == "X-RateLimit-Reset"
+    assert rest.limit_header == "X-RateLimit-Limit"
+    assert rest.exhaustion_status_codes == [429]
+    assert rest.is_response_aware
+    assert not graphql.is_response_aware
+
+
+def test_factory_does_not_share_instances_across_differing_header_config():
+    """Header config is part of the authenticator's behaviour, so it must take part in the
+    instance cache key -- otherwise two differently configured pools would silently share state."""
+    factory = ModelToComponentFactory()
+    config = {"pat": "token_1,token_2"}
+
+    plain = factory.create_rate_limited_multiple_token_authenticator(
+        _model("{{ config['pat'] }}"), config
+    )
+    response_aware_model = _model("{{ config['pat'] }}")
+    response_aware_model.quotas[0].remaining_header = "X-RateLimit-Remaining"
+    response_aware = factory.create_rate_limited_multiple_token_authenticator(
+        response_aware_model, config
+    )
+
+    assert plain is not response_aware
diff --git a/unit_tests/sources/streams/http/test_http_client.py b/unit_tests/sources/streams/http/test_http_client.py
index 48d396cb63..0b8b53471d 100644
--- a/unit_tests/sources/streams/http/test_http_client.py
+++ b/unit_tests/sources/streams/http/test_http_client.py
@@ -1062,6 +1062,75 @@ def update_response(*args, **kwargs):
     assert call_count == 2
 
 
+class _RecordingAuthenticator(TokenAuthenticator):
+    """An authenticator that tracks quota state and wants to see responses."""
+
+    def __init__(self):
+        super().__init__(token="token")
+        self.seen = []
+
+    def update_from_response(self, request, response):
+        self.seen.append(response.status_code)
+
+
+def test_authenticator_receives_every_response(requests_mock):
+    """A quota-tracking authenticator only ever sees requests, so `HttpClient` hands it the
+    responses too. Every attempt counts, including the retried ones."""
+    authenticator = _RecordingAuthenticator()
+    http_client = HttpClient(
+        name="test",
+        logger=logging.getLogger("test"),
+        authenticator=authenticator,
+        error_handler=HttpStatusErrorHandler(logger=logging.getLogger("test"), max_retries=1),
+    )
+    requests_mock.get(
+        "https://example.com/",
+        [{"status_code": 500}, {"status_code": 200}],
+    )
+
+    with patch("time.sleep"):
+        http_client.send_request(http_method="GET", url="https://example.com/", request_kwargs={})
+
+    assert authenticator.seen == [500, 200]
+
+
+def test_authenticator_without_update_hook_is_left_alone(requests_mock):
+    """Duck typed, so authenticators that don't track quota are untouched."""
+    http_client = HttpClient(
+        name="test",
+        logger=logging.getLogger("test"),
+        authenticator=TokenAuthenticator(token="token"),
+    )
+    requests_mock.get("https://example.com/", status_code=200)
+
+    _, response = http_client.send_request(
+        http_method="GET", url="https://example.com/", request_kwargs={}
+    )
+
+    assert response.status_code == 200
+
+
+def test_authenticator_update_failure_does_not_break_the_request(requests_mock):
+    """Quota bookkeeping is best-effort; it must never turn a good response into a failure."""
+
+    class _BrokenAuthenticator(TokenAuthenticator):
+        def update_from_response(self, request, response):
+            raise ValueError("boom")
+
+    http_client = HttpClient(
+        name="test",
+        logger=logging.getLogger("test"),
+        authenticator=_BrokenAuthenticator(token="token"),
+    )
+    requests_mock.get("https://example.com/", status_code=200, json={"ok": True})
+
+    _, response = http_client.send_request(
+        http_method="GET", url="https://example.com/", request_kwargs={}
+    )
+
+    assert response.json() == {"ok": True}
+
+
 def test_deprecated_alias_message_representation_airbyte_traced_errors_is_importable():
     """Verify that the deprecated alias still resolves to AirbyteTracedException."""
     assert MessageRepresentationAirbyteTracedErrors is AirbyteTracedException

From 51db59d496cecfbb3f5d6d9d73a71dda455cde51 Mon Sep 17 00:00:00 2001
From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com>
Date: Fri, 14 Aug 2026 17:37:48 +0300
Subject: [PATCH 02/14] fix(low-code): address review on response-aware token
 quota

Applies the PR #1117 review, plus one defect the review surfaced indirectly.

Not in the review, found while reproducing its F2: an exhaustion signal whose reset
header was *older* than the stored reset was dropped entirely. The window comparison
had three arms -- later, equal, earlier -- and the earlier arm discarded the whole
update, `remaining: 0` included. A 429 carrying a reset a few seconds behind the
seeded one therefore left the counter untouched and no rotation happened, which
silently disabled the feature for exactly the case it exists to handle. Measured
before the fix: six attempts on the same token, its counter ending at 4994 rather
than 0. The rule is now two-armed -- a later reset rolls the window over and takes
the server's numbers wholesale, anything else clamps with `min()` and never moves
the window backwards. Clamping unconditionally means no reset value can mask an
exhaustion signal, and dropping the third arm also resolves the review's F4 (a limit
from an older window overwriting the reserve), since there is no longer a class of
response that is accepted for one field and ignored for another.

F1: a pool zeroed from a response stayed out of rotation until every other token was
also spent, because rotation only skips spent tokens and the all-exhausted reseed is
the only path that refills them. `_acquire_call` now restores a pool whose window has
elapsed, and the all-exhausted test treats such a pool as available rather than spent,
so the sync picks the token up instead of dropping into the exhaustion wait and a
reseed round trip. The severity is lower than the review suggests: rotation is
"use one token until spent, then move on" rather than round-robin, so a parked token
costs no throughput while another has quota -- the cost is the needless wait when the
last one runs out. A `stale_zero` flag keeps the existing guard intact: when the quota
endpoint itself reports a pool spent with a reset already in the past, that is a
contradictory answer and must not be turned into invented quota.

F3: `is_response_aware` now includes `reset_header`, so a pool configured with it is
no longer silently inert, and a response proving the window rolled over restores the
pool to its limit even without a remaining header.

F6: a failing update hook warns once before falling back to debug, so a persistently
broken update is diagnosable rather than a silent reversion to single-token
behaviour. Added a `ResponseAwareAuthenticator` Protocol so the contract is
discoverable instead of implicit in a `getattr`.

F7: the threaded test asserted `remaining <= 4000`, which passes even if every local
decrement is lost. It now clamps once and then charges a known number of calls with
no further updates, asserting the exact expected count.

F8: dropped a test line guarding an `update_token` attribute that does not exist.

F2 is confirmed and needs no code change: the request that receives the rate limit
still pays its backoff, because the error handler computes that from the response
before rotation can matter. Only the retry moves to another token. The PR description
is corrected to say so, and an end-to-end test through HttpClient now pins the
rotation-on-retry behaviour, which until now rested on an uncovered re-sign in
`_send`.

F5 (field naming) is answered in the PR description rather than changed: within a
TokenQuota that already has `remaining_path`/`reset_path`/`limit_path`, the matching
`*_header` names read better than HTTPAPIBudget's `ratelimit_*` prefix.

Testing: 1530 tests pass across auth, parsers, requesters and streams.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../auth/rate_limited_multiple_token.py       |  68 ++++++--
 .../declarative_component_schema.yaml         |   2 +-
 .../models/declarative_component_schema.py    |   2 +-
 .../sources/streams/http/http_client.py       |  50 +++++-
 .../auth/test_rate_limited_multiple_token.py  | 149 +++++++++++++++---
 .../sources/streams/http/test_http_client.py  |  95 +++++++++++
 6 files changed, 330 insertions(+), 36 deletions(-)

diff --git a/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py b/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py
index ff95bce176..12ece3d730 100644
--- a/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py
+++ b/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py
@@ -48,7 +48,12 @@ class TokenQuota:
 
     @property
     def is_response_aware(self) -> bool:
-        return bool(self.remaining_header or self.limit_header or self.exhaustion_status_codes)
+        return bool(
+            self.remaining_header
+            or self.reset_header
+            or self.limit_header
+            or self.exhaustion_status_codes
+        )
 
 
 @dataclass
@@ -56,6 +61,11 @@ class _QuotaState:
     remaining: int
     reset_at: AirbyteDateTime
     limit: int
+    # True when the quota status endpoint reported this pool spent with a reset that had
+    # *already* passed -- a self-contradictory answer (server clock skew, or a lagging quota
+    # endpoint). Such a pool must not be restored from its elapsed window, or the connector
+    # would invent quota the server has just denied and spin on it.
+    stale_zero: bool = False
 
 
 class RateLimitedMultipleTokenAuthenticator(DeclarativeAuthenticator):
@@ -197,10 +207,13 @@ def _acquire_call(self, quota: TokenQuota) -> str:
             with self._lock:
                 token = self._active_token
                 state = self._states[token][quota.name]
+                self._restore_if_window_elapsed(state)
                 if state.remaining > 0:
                     state.remaining -= 1
                     budget_delay = self._compute_budget_delay(quota)
-                elif all(self._states[token][quota.name].remaining <= 0 for token in self._tokens):
+                elif all(
+                    self._is_pool_spent(self._states[token][quota.name]) for token in self._tokens
+                ):
                     now = time.monotonic()
                     if exhaustion_deadline is None:
                         exhaustion_deadline = now + self._max_wait_time.total_seconds()
@@ -246,6 +259,34 @@ def _acquire_call(self, quota: TokenQuota) -> str:
                 time.sleep(budget_delay)
             return token
 
+    @staticmethod
+    def _is_pool_spent(state: _QuotaState) -> bool:
+        """Whether a pool is out of calls with no prospect of recovering on its own.
+
+        A pool whose window has already elapsed is not spent -- its allowance is due back, so
+        the caller should pick it up rather than treat every token as exhausted and wait.
+        """
+        if state.remaining > 0:
+            return False
+        return state.stale_zero or state.reset_at > ab_datetime_now()
+
+    @staticmethod
+    def _restore_if_window_elapsed(state: _QuotaState) -> None:
+        """Give a spent pool its allowance back once its quota window has passed.
+
+        Rotation skips spent tokens and only the all-exhausted branch refills them, so a token
+        that reached zero stays out until *every* token is also at zero. Reading quota from
+        responses makes a pool reachable zero in a single rate-limited response, which would
+        otherwise send the sync into the exhaustion wait even when a token's window has already
+        rolled over and its calls are available again.
+
+        Must be called while holding the lock.
+        """
+        if state.stale_zero:
+            return
+        if state.remaining <= 0 and state.reset_at <= ab_datetime_now():
+            state.remaining = state.limit
+
     def _compute_budget_delay(self, quota: TokenQuota) -> Optional[float]:
         """Compute the proactive throttling delay. Must be called while holding the lock."""
         states = [self._states[token][quota.name] for token in self._tokens]
@@ -318,10 +359,12 @@ def _fetch_quota_states(self, token: str) -> dict[str, _QuotaState]:
                 if quota.limit_path
                 else remaining
             )
+            reset_at = ab_datetime_parse(reset)
             states[quota.name] = _QuotaState(
                 remaining=int(remaining),
-                reset_at=ab_datetime_parse(reset),
+                reset_at=reset_at,
                 limit=int(limit),
+                stale_zero=int(remaining) <= 0 and reset_at <= ab_datetime_now(),
             )
         return states
 
@@ -363,7 +406,7 @@ def update_from_response(
             remaining = 0
         reset_at = self._header_datetime(response, quota.reset_header)
         limit = self._header_int(response, quota.limit_header)
-        if remaining is None and limit is None:
+        if remaining is None and limit is None and reset_at is None:
             return
 
         with self._lock:
@@ -374,14 +417,17 @@ def update_from_response(
                 state.limit = limit
             if reset_at is not None and reset_at > state.reset_at:
                 # The quota window rolled over, so the local count is meaningless -- take the
-                # server's numbers wholesale.
+                # server's numbers wholesale. With no remaining header to go on, a fresh window
+                # is worth a full limit.
                 state.reset_at = reset_at
-                if remaining is not None:
-                    state.remaining = remaining
-            elif remaining is not None and (reset_at is None or reset_at == state.reset_at):
-                # Same window. Responses can arrive out of order and a slow one carries a stale,
-                # higher count, so only ever tighten the estimate -- never hand back calls that
-                # concurrent requests have already spent.
+                state.remaining = remaining if remaining is not None else state.limit
+            elif remaining is not None:
+                # Same window, or a response whose reset is older than what we hold. Only ever
+                # tighten the estimate: responses arrive out of order and a slow one carries a
+                # stale, higher count, so handing those calls back would let concurrent requests
+                # overspend. Clamping unconditionally also means an exhaustion signal is never
+                # dropped -- an earlier reset value must not be able to mask `remaining: 0`.
+                # The window itself is never moved backwards.
                 state.remaining = min(state.remaining, remaining)
 
     def _token_from_request(self, request: requests.PreparedRequest) -> Optional[str]:
diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml
index 06bfed587c..7de66bcf7f 100644
--- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml
+++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml
@@ -505,7 +505,7 @@ definitions:
           - "X-RateLimit-Remaining"
       reset_header:
         title: Reset Header
-        description: Optional response header carrying the quota reset timestamp for this pool. Parsed with the same rules as `reset_path`, so epoch seconds and ISO 8601 both work. Used to tell a rolled-over quota window from the current one.
+        description: Optional response header carrying the quota reset timestamp for this pool. Parsed with the same rules as `reset_path`, so epoch seconds and ISO 8601 both work. Used to tell a rolled-over quota window from the current one; a response proving the window has rolled over restores the pool to its limit. Most useful alongside `remaining_header`.
         type: string
         examples:
           - "X-RateLimit-Reset"
diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py
index f1b4c225b3..69ad3dcfe9 100644
--- a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py
+++ b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py
@@ -617,7 +617,7 @@ class TokenQuota(BaseModel):
     )
     reset_header: Optional[str] = Field(
         None,
-        description="Optional response header carrying the quota reset timestamp for this pool. Parsed with the same rules as `reset_path`, so epoch seconds and ISO 8601 both work. Used to tell a rolled-over quota window from the current one.",
+        description="Optional response header carrying the quota reset timestamp for this pool. Parsed with the same rules as `reset_path`, so epoch seconds and ISO 8601 both work. Used to tell a rolled-over quota window from the current one; a response proving the window has rolled over restores the pool to its limit. Most useful alongside `remaining_header`.",
         examples=["X-RateLimit-Reset"],
         title="Reset Header",
     )
diff --git a/airbyte_cdk/sources/streams/http/http_client.py b/airbyte_cdk/sources/streams/http/http_client.py
index 9b21a12af4..2eb2361aa5 100644
--- a/airbyte_cdk/sources/streams/http/http_client.py
+++ b/airbyte_cdk/sources/streams/http/http_client.py
@@ -6,7 +6,18 @@
 import os
 import urllib
 from pathlib import Path
-from typing import Any, Callable, Dict, List, Mapping, Optional, Tuple, Union
+from typing import (
+    Any,
+    Callable,
+    Dict,
+    List,
+    Mapping,
+    Optional,
+    Protocol,
+    Tuple,
+    Union,
+    runtime_checkable,
+)
 
 import orjson
 import requests
@@ -66,6 +77,22 @@
 BODY_REQUEST_METHODS = ("GET", "POST", "PUT", "PATCH")
 
 
+@runtime_checkable
+class ResponseAwareAuthenticator(Protocol):
+    """An authenticator that wants to see responses, not just sign requests.
+
+    Authenticators tracking per-token quota need a feedback channel: without one they can only
+    guess at the server's view of the quota and cannot tell that a token has been rejected.
+    `HttpClient` calls `update_from_response` once per attempt on any authenticator that
+    implements this, skipping replayed cache hits. Implementations must not raise for a response
+    they cannot interpret, and must be safe to call from multiple threads.
+    """
+
+    def update_from_response(
+        self, request: requests.PreparedRequest, response: requests.Response
+    ) -> None: ...
+
+
 def monkey_patched_get_item(self, key):  # type: ignore # this interface is a copy/paste from the requests_cache lib
     """
     con.execute can lead to `sqlite3.InterfaceError: bad parameter or other API misuse`. There was a fix implemented
@@ -138,6 +165,7 @@ def __init__(
         self._request_attempt_count: Dict[requests.PreparedRequest, int] = {}
         self._disable_retries = disable_retries
         self._message_repository = message_repository
+        self._authenticator_update_failed = False
 
     @property
     def cache_filename(self) -> str:
@@ -330,6 +358,8 @@ def _update_authenticator_from_response(
         Authenticators only ever see requests, so an authenticator that tracks per-token quota
         has no way to learn that the server disagrees with its local bookkeeping. This is the
         feedback channel, mirroring what `LimiterMixin.send` does for the API budget.
+
+        Any authenticator implementing `ResponseAwareAuthenticator` opts in.
         """
         authenticator = getattr(self._session, "auth", None)
         update_from_response = getattr(authenticator, "update_from_response", None)
@@ -342,10 +372,20 @@ def _update_authenticator_from_response(
         try:
             update_from_response(request, response)
         except Exception:
-            # Quota bookkeeping must never turn an otherwise fine response into a failure.
-            self._logger.debug(
-                "Authenticator failed to update quota state from response", exc_info=True
-            )
+            # Quota bookkeeping must never turn an otherwise fine response into a failure. Warn
+            # once so a persistently broken update -- which silently degrades the connector back
+            # to single-token behaviour -- is at least diagnosable from default-level logs.
+            if not self._authenticator_update_failed:
+                self._authenticator_update_failed = True
+                self._logger.warning(
+                    "Authenticator failed to update quota state from a response; token rotation "
+                    "may fall back to local counters only. Further occurrences log at debug.",
+                    exc_info=True,
+                )
+            else:
+                self._logger.debug(
+                    "Authenticator failed to update quota state from response", exc_info=True
+                )
 
     def _send(
         self,
diff --git a/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py b/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py
index 5220d534be..f34aee0144 100644
--- a/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py
+++ b/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py
@@ -26,6 +26,7 @@
 )
 from airbyte_cdk.sources.streams.call_rate import HttpRequestRegexMatcher
 from airbyte_cdk.utils import AirbyteTracedException
+from airbyte_cdk.utils.datetime_helpers import ab_datetime_now
 
 QUOTA_STATUS_URL = "https://api.example.com/rate_limit"
 
@@ -377,23 +378,25 @@ def test_later_reset_in_response_starts_a_new_window(requests_mock):
     assert int(state.reset_at.timestamp()) == next_window
 
 
-def test_earlier_reset_in_response_is_ignored(requests_mock):
-    """A response from a window that has already rolled over must not disturb current state."""
+def test_earlier_reset_in_response_never_moves_the_window_backwards(requests_mock):
+    """A response carrying an older reset still clamps the count -- dropping it outright would
+    let a stale reset value mask an exhaustion signal -- but it must not rewind the window."""
     requests_mock.get(QUOTA_STATUS_URL, json=_quota_status_body())
     authenticator = _response_aware_authenticator(tokens=("token_1",))
     request = authenticator(_prepared_request())
     state = authenticator._states["token_1"]["rest"]
     state.remaining = 100
-    previous_window = int(state.reset_at.timestamp()) - 3600
+    current_window = int(state.reset_at.timestamp())
 
     authenticator.update_from_response(
         request,
         _response(
-            headers={"X-RateLimit-Remaining": "7", "X-RateLimit-Reset": str(previous_window)}
+            headers={"X-RateLimit-Remaining": "7", "X-RateLimit-Reset": str(current_window - 3600)}
         ),
     )
 
-    assert state.remaining == 100
+    assert state.remaining == 7
+    assert int(state.reset_at.timestamp()) == current_window
 
 
 def test_limit_header_updates_the_throttling_reserve(requests_mock):
@@ -426,6 +429,105 @@ def test_exhaustion_status_code_zeroes_the_pool_and_next_request_rotates(request
     assert retry.headers["Authorization"] == "token token_2"
 
 
+def test_exhaustion_is_not_masked_by_an_earlier_reset_header(requests_mock):
+    """A rate-limited response must zero the pool even when its reset header is older than the
+    one we hold. Dropping the whole update in that case would silently disable rotation, since
+    the seeding read and the response headers routinely disagree by a few seconds."""
+    requests_mock.get(QUOTA_STATUS_URL, json=_quota_status_body(reset_in_seconds=3600))
+    authenticator = _response_aware_authenticator()
+    request = authenticator(_prepared_request())
+    earlier = int(authenticator._states["token_1"]["rest"].reset_at.timestamp()) - 1800
+
+    authenticator.update_from_response(
+        request,
+        _response(
+            status_code=429,
+            headers={"X-RateLimit-Remaining": "0", "X-RateLimit-Reset": str(earlier)},
+        ),
+    )
+
+    assert authenticator._states["token_1"]["rest"].remaining == 0
+    # the window itself is never moved backwards
+    assert int(authenticator._states["token_1"]["rest"].reset_at.timestamp()) == earlier + 1800
+    assert authenticator(_prepared_request()).headers["Authorization"] == "token token_2"
+
+
+def test_token_whose_window_elapsed_is_picked_up_instead_of_waiting(requests_mock):
+    """A pool zeroed from a response must not be treated as exhausted once its window rolls over.
+
+    Rotation skips spent tokens and only the all-exhausted branch refills them, so without this
+    the sync drops into the exhaustion wait (and a full reseed round trip) even though one
+    token's calls are already due back. Reading quota from responses makes a pool reachable
+    zero in a single response, which is what makes this easy to hit.
+    """
+    rate_limit_mock = requests_mock.get(QUOTA_STATUS_URL, json=_quota_status_body())
+    authenticator = _response_aware_authenticator()
+    request = authenticator(_prepared_request())
+    authenticator.update_from_response(request, _response(status_code=429))
+    assert authenticator._states["token_1"]["rest"].remaining == 0
+    # token_2 takes over and is spent in turn, while token_1's window rolls over
+    assert authenticator(_prepared_request()).headers["Authorization"] == "token token_2"
+    authenticator._states["token_2"]["rest"].remaining = 0
+    authenticator._states["token_1"]["rest"].reset_at = ab_datetime_now() - timedelta(seconds=1)
+    seeding_calls = rate_limit_mock.call_count
+
+    with patch("time.sleep") as mock_sleep:
+        signed = authenticator(_prepared_request())
+
+    assert signed.headers["Authorization"] == "token token_1"
+    assert not mock_sleep.called, "a token with an elapsed window must not trigger a wait"
+    assert rate_limit_mock.call_count == seeding_calls, "and must not need a reseed either"
+
+
+def test_pool_the_server_reports_spent_with_an_elapsed_reset_is_not_restored(requests_mock):
+    """Guard on the recovery above: when the quota endpoint itself answers "nothing left" with a
+    reset already in the past, that is a contradictory answer (clock skew, lagging endpoint).
+    Inventing quota from it would spin against a server that has just refused the call."""
+    requests_mock.get(
+        QUOTA_STATUS_URL, json=_quota_status_body(rest_remaining=0, reset_in_seconds=-10)
+    )
+    authenticator = RateLimitedMultipleTokenAuthenticator(
+        tokens=["token_1", "token_2"],
+        quotas=_response_aware_quotas(),
+        quota_status_url=QUOTA_STATUS_URL,
+        auth_method="token",
+        max_wait_time=timedelta(seconds=30),
+    )
+    authenticator._ensure_initialized()
+    assert all(authenticator._states[token]["rest"].stale_zero for token in authenticator._tokens)
+
+    # The exhaustion wait is bounded by wall clock, so the fake clock has to advance with the
+    # mocked sleeps or the reseed loop never reaches its deadline.
+    clock = {"now": 0.0}
+    with (
+        patch(
+            "time.sleep",
+            side_effect=lambda seconds: clock.__setitem__("now", clock["now"] + seconds),
+        ),
+        patch("time.monotonic", side_effect=lambda: clock["now"]),
+    ):
+        with pytest.raises(AirbyteTracedException, match="Rate limit is exceeded"):
+            authenticator(_prepared_request())
+
+
+def test_reset_only_response_restores_the_pool_on_a_new_window(requests_mock):
+    """A response that proves the window rolled over is actionable on its own: the previous
+    count belongs to a window that no longer exists."""
+    requests_mock.get(QUOTA_STATUS_URL, json=_quota_status_body())
+    authenticator = _response_aware_authenticator(tokens=("token_1",))
+    request = authenticator(_prepared_request())
+    state = authenticator._states["token_1"]["rest"]
+    state.remaining = 0
+    next_window = int(state.reset_at.timestamp()) + 3600
+
+    authenticator.update_from_response(
+        request, _response(headers={"X-RateLimit-Reset": str(next_window)})
+    )
+
+    assert state.remaining == state.limit
+    assert int(state.reset_at.timestamp()) == next_window
+
+
 def test_rate_limited_response_is_ignored_when_no_headers_are_configured(requests_mock):
     """Pools configured only with paths keep their previous behaviour."""
     requests_mock.get(QUOTA_STATUS_URL, json=_quota_status_body())
@@ -447,8 +549,9 @@ def test_response_is_attributed_to_the_token_that_sent_it(requests_mock):
     authenticator = _response_aware_authenticator()
     request = authenticator(_prepared_request())
     assert request.headers["Authorization"] == "token token_1"
-    authenticator.update_token = None  # guard against accidental reliance on a rotation helper
-    authenticator._active_token = "token_2"
+    authenticator._active_token = (
+        "token_2"  # the authenticator rotated while the call was in flight
+    )
 
     authenticator.update_from_response(request, _response(headers={"X-RateLimit-Remaining": "12"}))
 
@@ -499,29 +602,39 @@ def test_malformed_headers_are_ignored(requests_mock):
 
 
 def test_concurrent_response_updates_do_not_lose_decrements(requests_mock):
+    """Clamp once, then charge a known number of calls with no further updates, so the assertion
+    fails if a single decrement is lost rather than only if the clamp itself is."""
     requests_mock.get(QUOTA_STATUS_URL, json=_quota_status_body())
     authenticator = _response_aware_authenticator(tokens=("token_1",))
     authenticator._ensure_initialized()
     reset_at = authenticator._states["token_1"]["rest"].reset_at
-    headers = {
-        "X-RateLimit-Remaining": "4000",
-        "X-RateLimit-Reset": str(int(reset_at.timestamp())),
-    }
+    seed_request = authenticator(_prepared_request())
+    authenticator.update_from_response(
+        seed_request,
+        _response(
+            headers={
+                "X-RateLimit-Remaining": "4000",
+                "X-RateLimit-Reset": str(int(reset_at.timestamp())),
+            }
+        ),
+    )
+    assert authenticator._states["token_1"]["rest"].remaining == 4000
+
+    calls_per_thread, thread_count = 50, 4
 
     def worker():
-        for _ in range(50):
-            request = authenticator(_prepared_request())
-            authenticator.update_from_response(request, _response(headers=headers))
+        for _ in range(calls_per_thread):
+            authenticator(_prepared_request())
 
-    threads = [threading.Thread(target=worker) for _ in range(4)]
+    threads = [threading.Thread(target=worker) for _ in range(thread_count)]
     for thread in threads:
         thread.start()
     for thread in threads:
         thread.join()
 
-    # 200 requests were charged; the server's 4000 clamps the count and further local decrements
-    # continue from there, so the result must never exceed the clamp.
-    assert authenticator._states["token_1"]["rest"].remaining <= 4000
+    assert (
+        authenticator._states["token_1"]["rest"].remaining == 4000 - calls_per_thread * thread_count
+    )
 
 
 def _model(tokens):
diff --git a/unit_tests/sources/streams/http/test_http_client.py b/unit_tests/sources/streams/http/test_http_client.py
index 0b8b53471d..c4cf4abd02 100644
--- a/unit_tests/sources/streams/http/test_http_client.py
+++ b/unit_tests/sources/streams/http/test_http_client.py
@@ -2,6 +2,7 @@
 
 import logging
 import os
+import time
 from datetime import timedelta
 from unittest.mock import MagicMock, patch
 
@@ -1131,6 +1132,100 @@ def update_from_response(self, request, response):
     assert response.json() == {"ok": True}
 
 
+def test_rate_limited_token_rotates_on_the_retry_end_to_end(requests_mock):
+    """End-to-end: a rate-limited response zeroes that token's pool and the retry goes out on
+    another token.
+
+    This depends on `_send` re-signing the request on every attempt past the first. Nothing else
+    covers that line, so a refactor could silently drop rotation-on-retry and leave the sync
+    hammering a token the server has already rejected.
+
+    The residual sleep is the point of the assertion on `sleeps`: zeroing the pool does not
+    shorten the backoff for the request that *received* the rate limit -- the error handler has
+    already computed it from the response headers. The win is that the retry, and every request
+    after it, uses a token with quota.
+    """
+    from airbyte_cdk.sources.declarative.auth.rate_limited_multiple_token import (
+        RateLimitedMultipleTokenAuthenticator,
+        TokenQuota,
+    )
+    from airbyte_cdk.sources.declarative.requesters.error_handlers import DefaultErrorHandler
+    from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies import (
+        WaitUntilTimeFromHeaderBackoffStrategy,
+    )
+
+    quota_status_url = "https://api.example.com/rate_limit"
+    reset = int(time.time()) + 1800
+    requests_mock.get(
+        quota_status_url,
+        json={"resources": {"core": {"remaining": 5000, "reset": reset, "limit": 5000}}},
+    )
+    authenticator = RateLimitedMultipleTokenAuthenticator(
+        tokens=["token_1", "token_2"],
+        quotas=[
+            TokenQuota(
+                name="rest",
+                remaining_path=["resources", "core", "remaining"],
+                reset_path=["resources", "core", "reset"],
+                limit_path=["resources", "core", "limit"],
+                remaining_header="X-RateLimit-Remaining",
+                reset_header="X-RateLimit-Reset",
+                exhaustion_status_codes=[429],
+            )
+        ],
+        quota_status_url=quota_status_url,
+        auth_method="token",
+    )
+
+    tokens_used = []
+
+    def respond(request, context):
+        token = request.headers["Authorization"].split()[-1]
+        tokens_used.append(token)
+        if token == "token_1":
+            context.status_code = 429
+            context.headers = {"X-RateLimit-Reset": str(reset), "X-RateLimit-Remaining": "0"}
+            return "{}"
+        context.status_code = 200
+        context.headers = {}
+        return "{}"
+
+    requests_mock.get("https://api.example.com/data", text=respond)
+    # The backoff strategies must be on the client: that is what `_handle_error_resolution`
+    # consults. Passing them only to the error handler falls through to a 1s default and hides
+    # the real wait.
+    http_client = HttpClient(
+        name="test",
+        logger=logging.getLogger("test"),
+        authenticator=authenticator,
+        error_handler=DefaultErrorHandler(config={}, parameters={}),
+        backoff_strategy=[
+            WaitUntilTimeFromHeaderBackoffStrategy(
+                header="X-RateLimit-Reset", config={}, parameters={}
+            )
+        ],
+    )
+
+    sleeps = []
+    with patch("time.sleep", side_effect=lambda seconds: sleeps.append(seconds)):
+        _, response = http_client.send_request(
+            http_method="GET", url="https://api.example.com/data", request_kwargs={}
+        )
+
+    assert response.status_code == 200
+    assert tokens_used == ["token_1", "token_2"]
+    assert authenticator._states["token_1"]["rest"].remaining == 0
+    # One backoff was still paid, by the request that received the 429.
+    assert len([s for s in sleeps if s > 1]) == 1
+
+    tokens_used.clear()
+    with patch("time.sleep", side_effect=lambda seconds: sleeps.append(seconds)):
+        http_client.send_request(
+            http_method="GET", url="https://api.example.com/data", request_kwargs={}
+        )
+    assert tokens_used == ["token_2"]  # the spent token is not tried again
+
+
 def test_deprecated_alias_message_representation_airbyte_traced_errors_is_importable():
     """Verify that the deprecated alias still resolves to AirbyteTracedException."""
     assert MessageRepresentationAirbyteTracedErrors is AirbyteTracedException

From 8979db92532dc185f6db83c438242391d367eae1 Mon Sep 17 00:00:00 2001
From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com>
Date: Fri, 14 Aug 2026 17:45:44 +0300
Subject: [PATCH 03/14] fix(low-code): address CodeRabbit review on
 response-aware token quota

Normalize `exhaustion_status_codes` in the authenticator instance cache key. The key
stored the raw model value, which is `None` when the field is omitted, while the
runtime TokenQuota normalizes it to `[]`. Two definitions differing only that way
behaved identically but keyed differently, so their streams stopped sharing quota
counters -- the exact failure this component exists to prevent.

Check the auth-method prefix in `_token_from_request` before slicing it off, instead
of relying on `_states` membership to reject a mangled value. The header may have
been written by something other than this authenticator.

Use `pass` rather than `...` for the `ResponseAwareAuthenticator` Protocol body, which
static analysis flags as a statement with no effect. Both forms appear in the codebase
(`connector.py:110` uses `...`, `cursor_types.py` uses `pass`); no functional change.

CodeRabbit's remaining nitpick, on the threaded test asserting `remaining <= 4000`,
was already addressed in 51db59d4 -- the test now clamps once and asserts an exact
count afterwards.

Testing: 1102 tests pass across auth, parsers and streams; ruff and mypy clean.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../auth/rate_limited_multiple_token.py       | 15 +++++++--
 .../parsers/model_to_component_factory.py     |  4 ++-
 .../sources/streams/http/http_client.py       |  3 +-
 .../auth/test_rate_limited_multiple_token.py  | 31 +++++++++++++++++++
 4 files changed, 49 insertions(+), 4 deletions(-)

diff --git a/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py b/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py
index 12ece3d730..76c8a2f031 100644
--- a/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py
+++ b/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py
@@ -431,11 +431,22 @@ def update_from_response(
                 state.remaining = min(state.remaining, remaining)
 
     def _token_from_request(self, request: requests.PreparedRequest) -> Optional[str]:
-        """Recover the token a request was signed with from its auth header."""
+        """Recover the token a request was signed with from its auth header.
+
+        The prefix is checked rather than assumed: the header may have been written by
+        something other than this authenticator, and slicing blindly would leave membership in
+        `_states` as the only thing standing between a mangled value and a wrong attribution.
+        """
         value = request.headers.get(self._header)
         if not value:
             return None
-        token = value[len(self._auth_method) :].strip() if self._auth_method else value.strip()
+        if self._auth_method:
+            prefix = f"{self._auth_method} "
+            if not value.startswith(prefix):
+                return None
+            token = value[len(prefix) :].strip()
+        else:
+            token = value.strip()
         return token if token in self._states else None
 
     @staticmethod
diff --git a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py
index d9d3ecfc8f..a96c406d8c 100644
--- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py
+++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py
@@ -4663,7 +4663,9 @@ def create_rate_limited_multiple_token_authenticator(
                 "remaining_header": quota_model.remaining_header,
                 "reset_header": quota_model.reset_header,
                 "limit_header": quota_model.limit_header,
-                "exhaustion_status_codes": quota_model.exhaustion_status_codes,
+                # Normalized the same way as the runtime TokenQuota below, so an omitted field
+                # and an explicit `[]` key identically and keep sharing one set of counters.
+                "exhaustion_status_codes": quota_model.exhaustion_status_codes or [],
                 "matchers": [
                     {
                         "method": matcher_model.method,
diff --git a/airbyte_cdk/sources/streams/http/http_client.py b/airbyte_cdk/sources/streams/http/http_client.py
index 2eb2361aa5..41fe13cc6e 100644
--- a/airbyte_cdk/sources/streams/http/http_client.py
+++ b/airbyte_cdk/sources/streams/http/http_client.py
@@ -90,7 +90,8 @@ class ResponseAwareAuthenticator(Protocol):
 
     def update_from_response(
         self, request: requests.PreparedRequest, response: requests.Response
-    ) -> None: ...
+    ) -> None:
+        pass
 
 
 def monkey_patched_get_item(self, key):  # type: ignore # this interface is a copy/paste from the requests_cache lib
diff --git a/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py b/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py
index f34aee0144..ba9daf90d8 100644
--- a/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py
+++ b/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py
@@ -588,6 +588,21 @@ def test_graphql_response_updates_only_the_graphql_pool(requests_mock):
     assert authenticator._states["token_1"]["rest"].remaining == 5000
 
 
+def test_response_signed_by_something_else_is_not_attributed_to_a_token(requests_mock):
+    """The auth header may have been written by another component. Slicing off the configured
+    prefix without checking it would leave `_states` membership as the only guard against
+    charging the wrong token."""
+    requests_mock.get(QUOTA_STATUS_URL, json=_quota_status_body())
+    authenticator = _response_aware_authenticator(tokens=("token_1",))
+    authenticator._ensure_initialized()
+    request = _prepared_request()
+    request.headers["Authorization"] = "Bearer token_1"  # different scheme
+
+    authenticator.update_from_response(request, _response(headers={"X-RateLimit-Remaining": "1"}))
+
+    assert authenticator._states["token_1"]["rest"].remaining == 5000
+
+
 def test_malformed_headers_are_ignored(requests_mock):
     requests_mock.get(QUOTA_STATUS_URL, json=_quota_status_body())
     authenticator = _response_aware_authenticator(tokens=("token_1",))
@@ -801,6 +816,22 @@ def test_factory_passes_response_header_fields_through():
     assert not graphql.is_response_aware
 
 
+def test_factory_shares_instances_when_exhaustion_codes_are_omitted_versus_empty():
+    """An omitted `exhaustion_status_codes` and an explicit `[]` behave identically at runtime,
+    so they must not key differently -- splitting the counters is the very failure this
+    component exists to avoid."""
+    factory = ModelToComponentFactory()
+    config = {"pat": "token_1,token_2"}
+
+    omitted = _model("{{ config['pat'] }}")
+    explicit_empty = _model("{{ config['pat'] }}")
+    explicit_empty.quotas[0].exhaustion_status_codes = []
+
+    assert factory.create_rate_limited_multiple_token_authenticator(
+        omitted, config
+    ) is factory.create_rate_limited_multiple_token_authenticator(explicit_empty, config)
+
+
 def test_factory_does_not_share_instances_across_differing_header_config():
     """Header config is part of the authenticator's behaviour, so it must take part in the
     instance cache key -- otherwise two differently configured pools would silently share state."""

From 15d99fb704e158c8f987999d9f5addc92a4709a8 Mon Sep 17 00:00:00 2001
From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com>
Date: Mon, 17 Aug 2026 11:52:43 +0300
Subject: [PATCH 04/14] fix(low-code): drop local quota restore, reseed from
 the server instead

Reverts the elapsed-window restore added while addressing F1 of the earlier review.
CodeRabbit flagged that `_restore_if_window_elapsed` refilled `remaining` without
advancing `reset_at`; the consequence is worse than that framing.

`_acquire_call` runs for *every* pool, not just response-aware ones, so the restore
applied to the path-only configuration every existing consumer uses. With a reset
timestamp in the past the pool refilled itself on each drain, never reaching the
all-exhausted branch and therefore never reseeding. Measured on a path-only pool
whose quota status reports 3 calls with an already-elapsed reset: 12 calls served
from a 3-call allowance, one single quota-status request, zero waits. That is
unlimited invented quota against a limit the connector had stopped tracking -- a
regression against pre-PR behaviour, introduced to save a five-second wait.

The suggested remedy of retaining the window duration and advancing `reset_at` is
not implementable from what the component observes: quota status and response
headers both carry reset *timestamps*, never the window length, so the duration
would have to be guessed. Guessing it low re-creates the same over-issuing.

So the mechanism goes, and the original semantics return: when every token is spent
the authenticator waits until the earliest reset (floored at MIN_EXHAUSTION_WAIT so
a stale timestamp cannot busy-loop the quota endpoint) and reseeds from the server,
which is the only party that actually knows what the new window holds. The same
probe now shows 4 quota-status requests and three 5s waits across those 12 calls.

This leaves the original F1 observation unfixed by design, and that is the right
call: rotation is "use a token until it is spent, then move on" rather than
round-robin, so a token sitting at zero costs no throughput while another has quota.
The real cost was one 5s wait plus a reseed when the last token ran out. Trading
that for invented quota was a bad deal.

The `stale_zero` flag introduced to protect the existing stale-reset guard goes with
it, since without the restore there is nothing to guard against, which also resolves
the outside-diff comment about never clearing that flag.

Added a test pinning the behaviour I got wrong: an elapsed window must reseed from
the quota endpoint rather than refill locally.

Also from the review: `_response_aware_authenticator` now forwards `**kwargs` to the
authenticator rather than the quota builder, matching `_authenticator`, and the
rollover test asserts the concrete refill value instead of comparing a field to
itself. The third nitpick (a header-differs case in the instance-sharing test) is
already covered by test_factory_does_not_share_instances_across_differing_header_config.

Testing: 1101 tests pass across auth, parsers and streams; ruff and mypy clean.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../auth/rate_limited_multiple_token.py       |  42 +------
 .../auth/test_rate_limited_multiple_token.py  | 104 ++++++++----------
 2 files changed, 45 insertions(+), 101 deletions(-)

diff --git a/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py b/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py
index 76c8a2f031..1d84532db5 100644
--- a/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py
+++ b/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py
@@ -61,11 +61,6 @@ class _QuotaState:
     remaining: int
     reset_at: AirbyteDateTime
     limit: int
-    # True when the quota status endpoint reported this pool spent with a reset that had
-    # *already* passed -- a self-contradictory answer (server clock skew, or a lagging quota
-    # endpoint). Such a pool must not be restored from its elapsed window, or the connector
-    # would invent quota the server has just denied and spin on it.
-    stale_zero: bool = False
 
 
 class RateLimitedMultipleTokenAuthenticator(DeclarativeAuthenticator):
@@ -207,13 +202,10 @@ def _acquire_call(self, quota: TokenQuota) -> str:
             with self._lock:
                 token = self._active_token
                 state = self._states[token][quota.name]
-                self._restore_if_window_elapsed(state)
                 if state.remaining > 0:
                     state.remaining -= 1
                     budget_delay = self._compute_budget_delay(quota)
-                elif all(
-                    self._is_pool_spent(self._states[token][quota.name]) for token in self._tokens
-                ):
+                elif all(self._states[token][quota.name].remaining <= 0 for token in self._tokens):
                     now = time.monotonic()
                     if exhaustion_deadline is None:
                         exhaustion_deadline = now + self._max_wait_time.total_seconds()
@@ -259,34 +251,6 @@ def _acquire_call(self, quota: TokenQuota) -> str:
                 time.sleep(budget_delay)
             return token
 
-    @staticmethod
-    def _is_pool_spent(state: _QuotaState) -> bool:
-        """Whether a pool is out of calls with no prospect of recovering on its own.
-
-        A pool whose window has already elapsed is not spent -- its allowance is due back, so
-        the caller should pick it up rather than treat every token as exhausted and wait.
-        """
-        if state.remaining > 0:
-            return False
-        return state.stale_zero or state.reset_at > ab_datetime_now()
-
-    @staticmethod
-    def _restore_if_window_elapsed(state: _QuotaState) -> None:
-        """Give a spent pool its allowance back once its quota window has passed.
-
-        Rotation skips spent tokens and only the all-exhausted branch refills them, so a token
-        that reached zero stays out until *every* token is also at zero. Reading quota from
-        responses makes a pool reachable zero in a single rate-limited response, which would
-        otherwise send the sync into the exhaustion wait even when a token's window has already
-        rolled over and its calls are available again.
-
-        Must be called while holding the lock.
-        """
-        if state.stale_zero:
-            return
-        if state.remaining <= 0 and state.reset_at <= ab_datetime_now():
-            state.remaining = state.limit
-
     def _compute_budget_delay(self, quota: TokenQuota) -> Optional[float]:
         """Compute the proactive throttling delay. Must be called while holding the lock."""
         states = [self._states[token][quota.name] for token in self._tokens]
@@ -359,12 +323,10 @@ def _fetch_quota_states(self, token: str) -> dict[str, _QuotaState]:
                 if quota.limit_path
                 else remaining
             )
-            reset_at = ab_datetime_parse(reset)
             states[quota.name] = _QuotaState(
                 remaining=int(remaining),
-                reset_at=reset_at,
+                reset_at=ab_datetime_parse(reset),
                 limit=int(limit),
-                stale_zero=int(remaining) <= 0 and reset_at <= ab_datetime_now(),
             )
         return states
 
diff --git a/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py b/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py
index ba9daf90d8..4fc3572fa1 100644
--- a/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py
+++ b/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py
@@ -148,6 +148,40 @@ def test_waits_and_reseeds_when_all_tokens_exhausted(requests_mock):
     )
 
 
+def test_elapsed_window_reseeds_from_the_server_rather_than_refilling_locally(requests_mock):
+    """An exhausted pool whose reset has passed must be refreshed from the quota endpoint.
+
+    Handing the allowance back locally instead would let the connector invent quota it has no
+    evidence for: the reset timestamp says a new window is due, but only the server knows what
+    is actually left in it. A pool refilled from a stale timestamp never reseeds and never
+    waits, so it would serve unlimited calls against a quota it stopped tracking.
+    """
+    body = {"resources": {"core": {"remaining": 3, "reset": int(time.time()) - 10, "limit": 3}}}
+    quota_status = requests_mock.get(QUOTA_STATUS_URL, json=body)
+    authenticator = RateLimitedMultipleTokenAuthenticator(
+        tokens=["token_1"],
+        quotas=[
+            TokenQuota(
+                name="rest",
+                remaining_path=["resources", "core", "remaining"],
+                reset_path=["resources", "core", "reset"],
+                limit_path=["resources", "core", "limit"],
+            )
+        ],
+        quota_status_url=QUOTA_STATUS_URL,
+        auth_method="token",
+    )
+
+    sleeps = []
+    with patch("time.sleep", side_effect=lambda seconds: sleeps.append(seconds)):
+        for _ in range(12):
+            authenticator(_prepared_request())
+
+    # 3 calls per window: the initial seeding plus one reseed for each exhausted window.
+    assert quota_status.call_count == 4
+    assert sleeps == [RateLimitedMultipleTokenAuthenticator.MIN_EXHAUSTION_WAIT] * 3
+
+
 def test_budget_throttling_delay_injected_when_all_tokens_below_reserve(requests_mock):
     requests_mock.get(QUOTA_STATUS_URL, json=_quota_status_body(rest_remaining=100))
     authenticator = _authenticator()
@@ -309,12 +343,18 @@ def _response_aware_quotas(exhaustion_status_codes=(429,)):
     ]
 
 
-def _response_aware_authenticator(tokens=("token_1", "token_2"), **kwargs):
+def _response_aware_authenticator(
+    tokens=("token_1", "token_2"), exhaustion_status_codes=(429,), **kwargs
+):
+    """`**kwargs` reach the authenticator, matching `_authenticator` above; quota-shaping
+    arguments are named explicitly so a caller passing e.g. `max_wait_time` doesn't hit a
+    TypeError from the quota builder."""
     return RateLimitedMultipleTokenAuthenticator(
         tokens=list(tokens),
-        quotas=_response_aware_quotas(**kwargs),
+        quotas=_response_aware_quotas(exhaustion_status_codes=exhaustion_status_codes),
         quota_status_url=QUOTA_STATUS_URL,
         auth_method="token",
+        **kwargs,
     )
 
 
@@ -452,64 +492,6 @@ def test_exhaustion_is_not_masked_by_an_earlier_reset_header(requests_mock):
     assert authenticator(_prepared_request()).headers["Authorization"] == "token token_2"
 
 
-def test_token_whose_window_elapsed_is_picked_up_instead_of_waiting(requests_mock):
-    """A pool zeroed from a response must not be treated as exhausted once its window rolls over.
-
-    Rotation skips spent tokens and only the all-exhausted branch refills them, so without this
-    the sync drops into the exhaustion wait (and a full reseed round trip) even though one
-    token's calls are already due back. Reading quota from responses makes a pool reachable
-    zero in a single response, which is what makes this easy to hit.
-    """
-    rate_limit_mock = requests_mock.get(QUOTA_STATUS_URL, json=_quota_status_body())
-    authenticator = _response_aware_authenticator()
-    request = authenticator(_prepared_request())
-    authenticator.update_from_response(request, _response(status_code=429))
-    assert authenticator._states["token_1"]["rest"].remaining == 0
-    # token_2 takes over and is spent in turn, while token_1's window rolls over
-    assert authenticator(_prepared_request()).headers["Authorization"] == "token token_2"
-    authenticator._states["token_2"]["rest"].remaining = 0
-    authenticator._states["token_1"]["rest"].reset_at = ab_datetime_now() - timedelta(seconds=1)
-    seeding_calls = rate_limit_mock.call_count
-
-    with patch("time.sleep") as mock_sleep:
-        signed = authenticator(_prepared_request())
-
-    assert signed.headers["Authorization"] == "token token_1"
-    assert not mock_sleep.called, "a token with an elapsed window must not trigger a wait"
-    assert rate_limit_mock.call_count == seeding_calls, "and must not need a reseed either"
-
-
-def test_pool_the_server_reports_spent_with_an_elapsed_reset_is_not_restored(requests_mock):
-    """Guard on the recovery above: when the quota endpoint itself answers "nothing left" with a
-    reset already in the past, that is a contradictory answer (clock skew, lagging endpoint).
-    Inventing quota from it would spin against a server that has just refused the call."""
-    requests_mock.get(
-        QUOTA_STATUS_URL, json=_quota_status_body(rest_remaining=0, reset_in_seconds=-10)
-    )
-    authenticator = RateLimitedMultipleTokenAuthenticator(
-        tokens=["token_1", "token_2"],
-        quotas=_response_aware_quotas(),
-        quota_status_url=QUOTA_STATUS_URL,
-        auth_method="token",
-        max_wait_time=timedelta(seconds=30),
-    )
-    authenticator._ensure_initialized()
-    assert all(authenticator._states[token]["rest"].stale_zero for token in authenticator._tokens)
-
-    # The exhaustion wait is bounded by wall clock, so the fake clock has to advance with the
-    # mocked sleeps or the reseed loop never reaches its deadline.
-    clock = {"now": 0.0}
-    with (
-        patch(
-            "time.sleep",
-            side_effect=lambda seconds: clock.__setitem__("now", clock["now"] + seconds),
-        ),
-        patch("time.monotonic", side_effect=lambda: clock["now"]),
-    ):
-        with pytest.raises(AirbyteTracedException, match="Rate limit is exceeded"):
-            authenticator(_prepared_request())
-
-
 def test_reset_only_response_restores_the_pool_on_a_new_window(requests_mock):
     """A response that proves the window rolled over is actionable on its own: the previous
     count belongs to a window that no longer exists."""
@@ -524,7 +506,7 @@ def test_reset_only_response_restores_the_pool_on_a_new_window(requests_mock):
         request, _response(headers={"X-RateLimit-Reset": str(next_window)})
     )
 
-    assert state.remaining == state.limit
+    assert state.remaining == 5000  # the pool's full limit, for the window just opened
     assert int(state.reset_at.timestamp()) == next_window
 
 

From 2d0953c01a7f1cf3db7fde44cd864dc3042c0390 Mon Sep 17 00:00:00 2001
From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com>
Date: Mon, 17 Aug 2026 12:05:17 +0300
Subject: [PATCH 05/14] feat(low-code): retry on another token instead of
 waiting out a rate limit

Closes the gap left open by the response-header work: a rate-limited request paid the
full backoff even when a second token sat completely unused. Probe at the moment of
the sleep, two tokens, the first rate-limited with a reset 30 minutes out:

  SLEEP 1800.3s | token state: {'token_1': 0, 'token_2': 5000}

The wait guarded nothing. It happens because the backoff is computed by the error
handler from the rejected response's headers, which describe one credential; the
authenticator, which is the only component that knows a spare exists, has no say.

`HttpClient` now asks. When a resolution is RATE_LIMITED and the authenticator reports
`has_alternative_token(request)`, the computed wait is replaced with a token-rotation
backoff and the retry -- which re-signs the request -- goes out on the other token.
Same probe now sleeps 1.1s.

`has_alternative_token` is deliberately narrow. It answers True only when the token
that *sent* the request is spent for the matched quota pool, guaranteeing the next
acquire rotates, and some other token is not. If the sending token still has calls the
rejection was not about exhausting it -- a secondary limit is typically per-account and
would reject every token alike -- so the computed wait stands. Availability is answered
per quota pool, so exhausting graphql does not shortcut a rest wait.

Three properties worth stating, all covered by tests:

- Only RATE_LIMITED is shortened. A 500 retry has nothing to do with credentials and
  keeps its backoff.
- The replacement backoff is non-zero, so an authenticator that misreports availability
  degrades to a slow retry rather than a hot loop against a live rate limit.
- Authenticators without the hook are untouched, as is any authenticator that reports
  no spare.

This is what the source-github Python streams already do in connector-specific code
(rotate, then return a 1s backoff instead of the computed one), so the declarative path
no longer regresses against the implementation it replaces -- which was the reason to
treat the CDK work as a prerequisite for migrating that connector's heavy streams.

Testing: 1110 tests pass across auth, parsers and streams; ruff and mypy clean.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../auth/rate_limited_multiple_token.py       |  27 +++++
 .../sources/streams/http/http_client.py       |  51 +++++++++
 .../auth/test_rate_limited_multiple_token.py  |  61 ++++++++++
 .../sources/streams/http/test_http_client.py  | 104 +++++++++++++++++-
 4 files changed, 237 insertions(+), 6 deletions(-)

diff --git a/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py b/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py
index 1d84532db5..2d4cbbd738 100644
--- a/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py
+++ b/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py
@@ -392,6 +392,33 @@ def update_from_response(
                 # The window itself is never moved backwards.
                 state.remaining = min(state.remaining, remaining)
 
+    def has_alternative_token(self, request: requests.PreparedRequest) -> bool:
+        """Whether another token could serve this request right now.
+
+        Answers the question a rate-limit backoff cannot answer for itself: the wait computed
+        from a response's reset header assumes the only way forward is for that quota to come
+        back, which is false when a different token still has calls. `HttpClient` uses this to
+        retry promptly instead of sleeping out a window it does not need.
+
+        Deliberately narrow. It reports True only when the token that *sent* the request is
+        spent for the matched pool -- so the next request is guaranteed to rotate -- and some
+        other token is not. If the sending token still has calls locally, the rejection was not
+        about exhausting it (a secondary limit, say, which on many APIs is per-user and would
+        reject every token alike), and waiting remains the right response.
+        """
+        quota = self._match_quota(request)
+        sender = self._token_from_request(request)
+        with self._lock:
+            if not self._states or sender is None:
+                return False
+            if self._states[sender][quota.name].remaining > 0:
+                return False
+            return any(
+                self._states[token][quota.name].remaining > 0
+                for token in self._tokens
+                if token != sender
+            )
+
     def _token_from_request(self, request: requests.PreparedRequest) -> Optional[str]:
         """Recover the token a request was signed with from its auth header.
 
diff --git a/airbyte_cdk/sources/streams/http/http_client.py b/airbyte_cdk/sources/streams/http/http_client.py
index 41fe13cc6e..4e9d17b983 100644
--- a/airbyte_cdk/sources/streams/http/http_client.py
+++ b/airbyte_cdk/sources/streams/http/http_client.py
@@ -94,6 +94,22 @@ def update_from_response(
         pass
 
 
+@runtime_checkable
+class TokenRotatingAuthenticator(Protocol):
+    """An authenticator holding several interchangeable credentials.
+
+    A rate-limit backoff is computed from the response alone, so it assumes the only way
+    forward is for that quota to come back. An authenticator with a spare credential knows
+    better. `HttpClient` asks before sleeping out a rate-limit window.
+
+    Implementations must only answer True when retrying immediately would actually use a
+    different credential -- otherwise the retry hammers the same rejected one.
+    """
+
+    def has_alternative_token(self, request: requests.PreparedRequest) -> bool:
+        pass
+
+
 def monkey_patched_get_item(self, key):  # type: ignore # this interface is a copy/paste from the requests_cache lib
     """
     con.execute can lead to `sqlite3.InterfaceError: bad parameter or other API misuse`. There was a fix implemented
@@ -118,6 +134,10 @@ def monkey_patched_get_item(self, key):  # type: ignore # this interface is a co
 class HttpClient:
     _DEFAULT_MAX_RETRY: int = 5
     _DEFAULT_MAX_TIME: int = 60 * 10
+    # Backoff used in place of a rate-limit wait when another credential can serve the retry.
+    # Kept non-zero so a misreporting authenticator degrades to a slow retry, not a hot loop
+    # (the retry handler adds a second on top of whatever is returned).
+    TOKEN_ROTATION_BACKOFF: float = 0.1
     _ACTIONS_TO_RETRY_ON = {
         ResponseAction.RETRY,
         ResponseAction.RATE_LIMITED,
@@ -351,6 +371,21 @@ def _send_with_retry(
                 stream_descriptor=StreamDescriptor(name=self._name),
             )
 
+    def _can_retry_on_another_token(self, request: requests.PreparedRequest) -> bool:
+        """Whether the authenticator can serve this request from a different credential now."""
+        authenticator = getattr(self._session, "auth", None)
+        has_alternative_token = getattr(authenticator, "has_alternative_token", None)
+        if not callable(has_alternative_token):
+            return False
+        try:
+            return bool(has_alternative_token(request))
+        except Exception:
+            # Falling back to the computed wait is always safe, so never fail a retry over this.
+            self._logger.debug(
+                "Authenticator failed to report credential availability", exc_info=True
+            )
+            return False
+
     def _update_authenticator_from_response(
         self, request: requests.PreparedRequest, response: requests.Response
     ) -> None:
@@ -596,6 +631,22 @@ def _handle_error_resolution(
                 if backoff_time:
                     user_defined_backoff_time = backoff_time
                     break
+
+            if (
+                user_defined_backoff_time
+                and error_resolution.response_action == ResponseAction.RATE_LIMITED
+                and self._can_retry_on_another_token(request)
+            ):
+                # The backoff was derived from this response's headers, which only describe the
+                # credential that was rejected. Another one has quota, and the retry re-signs the
+                # request, so waiting out this window would idle for nothing.
+                self._logger.info(
+                    f"Rate limited on the current credential; retrying in "
+                    f"{self.TOKEN_ROTATION_BACKOFF}s with another one instead of waiting "
+                    f"{user_defined_backoff_time:.0f}s for the rate limit to reset."
+                )
+                user_defined_backoff_time = self.TOKEN_ROTATION_BACKOFF
+
             error_message = (
                 error_resolution.error_message
                 or f"Request to {request.url} failed with failure type {error_resolution.failure_type}, response action {error_resolution.response_action}."
diff --git a/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py b/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py
index 4fc3572fa1..48fd6901a4 100644
--- a/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py
+++ b/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py
@@ -570,6 +570,67 @@ def test_graphql_response_updates_only_the_graphql_pool(requests_mock):
     assert authenticator._states["token_1"]["rest"].remaining == 5000
 
 
+def test_has_alternative_token_when_sender_is_spent_and_another_is_not(requests_mock):
+    requests_mock.get(QUOTA_STATUS_URL, json=_quota_status_body())
+    authenticator = _response_aware_authenticator()
+    request = authenticator(_prepared_request())
+    authenticator.update_from_response(request, _response(status_code=429))
+
+    assert authenticator.has_alternative_token(request) is True
+
+
+def test_no_alternative_token_while_the_sending_token_still_has_calls(requests_mock):
+    """A rejection that did not exhaust the sending token is not something rotation fixes --
+    a secondary limit is typically per-account and would reject every token alike, so the
+    computed wait must stand."""
+    requests_mock.get(QUOTA_STATUS_URL, json=_quota_status_body())
+    authenticator = _response_aware_authenticator()
+    request = authenticator(_prepared_request())
+
+    assert authenticator.has_alternative_token(request) is False
+
+
+def test_no_alternative_token_when_every_token_is_spent(requests_mock):
+    requests_mock.get(QUOTA_STATUS_URL, json=_quota_status_body())
+    authenticator = _response_aware_authenticator()
+    request = authenticator(_prepared_request())
+    authenticator._ensure_initialized()
+    for token in authenticator._tokens:
+        authenticator._states[token]["rest"].remaining = 0
+
+    assert authenticator.has_alternative_token(request) is False
+
+
+def test_no_alternative_token_with_a_single_token(requests_mock):
+    requests_mock.get(QUOTA_STATUS_URL, json=_quota_status_body())
+    authenticator = _response_aware_authenticator(tokens=("token_1",))
+    request = authenticator(_prepared_request())
+    authenticator.update_from_response(request, _response(status_code=429))
+
+    assert authenticator.has_alternative_token(request) is False
+
+
+def test_alternative_token_is_scoped_to_the_matched_quota_pool(requests_mock):
+    """Availability is answered per pool: exhausting graphql says nothing about rest.
+
+    The graphql pool here declares `remaining_header` but no `exhaustion_status_codes`, so it
+    learns it is spent from the header rather than from the status code.
+    """
+    requests_mock.get(QUOTA_STATUS_URL, json=_quota_status_body())
+    authenticator = _response_aware_authenticator()
+    graphql_request = authenticator(_prepared_request("https://api.example.com/graphql"))
+    rest_request = authenticator(_prepared_request())
+    authenticator.update_from_response(
+        graphql_request, _response(status_code=429, headers={"X-RateLimit-Remaining": "0"})
+    )
+
+    assert authenticator._states["token_1"]["graphql"].remaining == 0
+    assert authenticator.has_alternative_token(graphql_request) is True
+    # the rest pool is untouched, so nothing about it is "spent" and no rotation is implied
+    assert authenticator._states["token_1"]["rest"].remaining == 4999
+    assert authenticator.has_alternative_token(rest_request) is False
+
+
 def test_response_signed_by_something_else_is_not_attributed_to_a_token(requests_mock):
     """The auth header may have been written by another component. Slicing off the configured
     prefix without checking it would leave `_states` membership as the only guard against
diff --git a/unit_tests/sources/streams/http/test_http_client.py b/unit_tests/sources/streams/http/test_http_client.py
index c4cf4abd02..d7e0e51f81 100644
--- a/unit_tests/sources/streams/http/test_http_client.py
+++ b/unit_tests/sources/streams/http/test_http_client.py
@@ -1140,10 +1140,8 @@ def test_rate_limited_token_rotates_on_the_retry_end_to_end(requests_mock):
     covers that line, so a refactor could silently drop rotation-on-retry and leave the sync
     hammering a token the server has already rejected.
 
-    The residual sleep is the point of the assertion on `sleeps`: zeroing the pool does not
-    shorten the backoff for the request that *received* the rate limit -- the error handler has
-    already computed it from the response headers. The win is that the retry, and every request
-    after it, uses a token with quota.
+    The 30-minute reset in the response is deliberately far away: the assertion on `sleeps` is
+    that it is *not* waited out, because the authenticator reports another token with quota.
     """
     from airbyte_cdk.sources.declarative.auth.rate_limited_multiple_token import (
         RateLimitedMultipleTokenAuthenticator,
@@ -1215,8 +1213,8 @@ def respond(request, context):
     assert response.status_code == 200
     assert tokens_used == ["token_1", "token_2"]
     assert authenticator._states["token_1"]["rest"].remaining == 0
-    # One backoff was still paid, by the request that received the 429.
-    assert len([s for s in sleeps if s > 1]) == 1
+    # The 1800s reset is skipped: a token with quota was available the whole time.
+    assert max(sleeps) < 5, f"expected a prompt retry, slept {sleeps}"
 
     tokens_used.clear()
     with patch("time.sleep", side_effect=lambda seconds: sleeps.append(seconds)):
@@ -1226,6 +1224,100 @@ def respond(request, context):
     assert tokens_used == ["token_2"]  # the spent token is not tried again
 
 
+class _SpareTokenAuthenticator(TokenAuthenticator):
+    """Reports a spare credential without tracking any quota."""
+
+    def __init__(self, has_spare=True):
+        super().__init__(token="token")
+        self.has_spare = has_spare
+
+    def has_alternative_token(self, request):
+        return self.has_spare
+
+
+def _rate_limited_client(authenticator, backoff_seconds=1800):
+    return HttpClient(
+        name="test",
+        logger=logging.getLogger("test"),
+        authenticator=authenticator,
+        error_handler=HttpStatusErrorHandler(
+            logger=logging.getLogger("test"),
+            error_mapping={
+                429: ErrorResolution(
+                    response_action=ResponseAction.RATE_LIMITED,
+                    failure_type=FailureType.transient_error,
+                    error_message="rate limited",
+                ),
+                500: ErrorResolution(
+                    response_action=ResponseAction.RETRY,
+                    failure_type=FailureType.transient_error,
+                    error_message="server error",
+                ),
+            },
+            max_retries=1,
+        ),
+        backoff_strategy=_ConstantBackoffStrategy(backoff_seconds),
+    )
+
+
+class _ConstantBackoffStrategy(BackoffStrategy):
+    def __init__(self, seconds):
+        self.seconds = seconds
+
+    def backoff_time(self, *args, **kwargs):
+        return self.seconds
+
+
+def test_rate_limit_wait_is_skipped_when_another_credential_is_available(requests_mock):
+    """The backoff is computed from the rejected credential's response, so it cannot know a
+    spare one is idle. The client asks before sleeping."""
+    requests_mock.get("https://example.com/", [{"status_code": 429}, {"status_code": 200}])
+    client = _rate_limited_client(_SpareTokenAuthenticator(has_spare=True))
+
+    sleeps = []
+    with patch("time.sleep", side_effect=lambda seconds: sleeps.append(seconds)):
+        _, response = client.send_request(
+            http_method="GET", url="https://example.com/", request_kwargs={}
+        )
+
+    assert response.status_code == 200
+    assert max(sleeps) < 5, f"expected a prompt retry, slept {sleeps}"
+
+
+def test_rate_limit_wait_is_paid_when_no_other_credential_is_available(requests_mock):
+    requests_mock.get("https://example.com/", [{"status_code": 429}, {"status_code": 200}])
+    client = _rate_limited_client(_SpareTokenAuthenticator(has_spare=False))
+
+    sleeps = []
+    with patch("time.sleep", side_effect=lambda seconds: sleeps.append(seconds)):
+        client.send_request(http_method="GET", url="https://example.com/", request_kwargs={})
+
+    assert max(sleeps) > 1000, f"the full rate-limit wait should stand, slept {sleeps}"
+
+
+def test_non_rate_limit_retry_is_not_shortened_by_a_spare_credential(requests_mock):
+    """A 500 has nothing to do with credentials; its backoff must be left alone."""
+    requests_mock.get("https://example.com/", [{"status_code": 500}, {"status_code": 200}])
+    client = _rate_limited_client(_SpareTokenAuthenticator(has_spare=True))
+
+    sleeps = []
+    with patch("time.sleep", side_effect=lambda seconds: sleeps.append(seconds)):
+        client.send_request(http_method="GET", url="https://example.com/", request_kwargs={})
+
+    assert max(sleeps) > 1000, f"a server-error backoff should stand, slept {sleeps}"
+
+
+def test_authenticator_without_alternative_token_hook_pays_the_full_wait(requests_mock):
+    requests_mock.get("https://example.com/", [{"status_code": 429}, {"status_code": 200}])
+    client = _rate_limited_client(TokenAuthenticator(token="token"))
+
+    sleeps = []
+    with patch("time.sleep", side_effect=lambda seconds: sleeps.append(seconds)):
+        client.send_request(http_method="GET", url="https://example.com/", request_kwargs={})
+
+    assert max(sleeps) > 1000
+
+
 def test_deprecated_alias_message_representation_airbyte_traced_errors_is_importable():
     """Verify that the deprecated alias still resolves to AirbyteTracedException."""
     assert MessageRepresentationAirbyteTracedErrors is AirbyteTracedException

From fc88df17d19899fedcd0441638ee52db33cbbc6b Mon Sep 17 00:00:00 2001
From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com>
Date: Mon, 17 Aug 2026 12:21:01 +0300
Subject: [PATCH 06/14] refactor(low-code): move authenticator capability
 protocols next to the authenticators

`ResponseAwareAuthenticator` and `TokenRotatingAuthenticator` described what an
authenticator can offer, but lived in `http_client.py` because that is what calls
them. Anyone writing an authenticator would look under `requests_native_auth/`,
which is where the base classes live and where `DeclarativeAuthenticator` inherits
from, so that is where the contracts belong.

Moved to `requests_native_auth/protocols.py` and exported from the package. This
costs nothing in the import graph: `HttpClient` dispatches structurally through
`getattr`, so it never referenced the protocol objects at runtime -- only a docstring
mentioned them. `streams/http` still imports nothing from `declarative`, and the
protocols' new home does not import `HttpClient`, so an authenticator that needs the
client (as the declarative one does) cannot create a cycle.

`RateLimitedMultipleTokenAuthenticator` now names the two protocols it implements, so
the contract is reachable from either end.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../auth/rate_limited_multiple_token.py       |  4 ++
 .../sources/streams/http/http_client.py       | 53 ++----------------
 .../http/requests_native_auth/__init__.py     |  3 +
 .../http/requests_native_auth/protocols.py    | 55 +++++++++++++++++++
 4 files changed, 68 insertions(+), 47 deletions(-)
 create mode 100644 airbyte_cdk/sources/streams/http/requests_native_auth/protocols.py

diff --git a/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py b/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py
index 2d4cbbd738..5b1e656675 100644
--- a/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py
+++ b/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py
@@ -77,6 +77,10 @@ class RateLimitedMultipleTokenAuthenticator(DeclarativeAuthenticator):
     (`max(budget_min_reserve, budget_reserve_fraction * limit)`), a small delay proportional to
     `seconds_until_reset / total_remaining` (capped at 10s) is injected before each request.
 
+    Implements `ResponseAwareAuthenticator` and `TokenRotatingAuthenticator` (see
+    `airbyte_cdk.sources.streams.http.requests_native_auth.protocols`), which is how `HttpClient`
+    feeds it responses and asks it whether a rate-limit wait can be skipped.
+
     Counters are seeded per token from `quota_status_url` on first use and refreshed after an
     exhaustion wait. When a pool declares response headers, `update_from_response` additionally
     reconciles that pool against the server on every response, which keeps the counters honest
diff --git a/airbyte_cdk/sources/streams/http/http_client.py b/airbyte_cdk/sources/streams/http/http_client.py
index 4e9d17b983..c23dfcb9e5 100644
--- a/airbyte_cdk/sources/streams/http/http_client.py
+++ b/airbyte_cdk/sources/streams/http/http_client.py
@@ -6,18 +6,7 @@
 import os
 import urllib
 from pathlib import Path
-from typing import (
-    Any,
-    Callable,
-    Dict,
-    List,
-    Mapping,
-    Optional,
-    Protocol,
-    Tuple,
-    Union,
-    runtime_checkable,
-)
+from typing import Any, Callable, Dict, List, Mapping, Optional, Tuple, Union
 
 import orjson
 import requests
@@ -77,39 +66,6 @@
 BODY_REQUEST_METHODS = ("GET", "POST", "PUT", "PATCH")
 
 
-@runtime_checkable
-class ResponseAwareAuthenticator(Protocol):
-    """An authenticator that wants to see responses, not just sign requests.
-
-    Authenticators tracking per-token quota need a feedback channel: without one they can only
-    guess at the server's view of the quota and cannot tell that a token has been rejected.
-    `HttpClient` calls `update_from_response` once per attempt on any authenticator that
-    implements this, skipping replayed cache hits. Implementations must not raise for a response
-    they cannot interpret, and must be safe to call from multiple threads.
-    """
-
-    def update_from_response(
-        self, request: requests.PreparedRequest, response: requests.Response
-    ) -> None:
-        pass
-
-
-@runtime_checkable
-class TokenRotatingAuthenticator(Protocol):
-    """An authenticator holding several interchangeable credentials.
-
-    A rate-limit backoff is computed from the response alone, so it assumes the only way
-    forward is for that quota to come back. An authenticator with a spare credential knows
-    better. `HttpClient` asks before sleeping out a rate-limit window.
-
-    Implementations must only answer True when retrying immediately would actually use a
-    different credential -- otherwise the retry hammers the same rejected one.
-    """
-
-    def has_alternative_token(self, request: requests.PreparedRequest) -> bool:
-        pass
-
-
 def monkey_patched_get_item(self, key):  # type: ignore # this interface is a copy/paste from the requests_cache lib
     """
     con.execute can lead to `sqlite3.InterfaceError: bad parameter or other API misuse`. There was a fix implemented
@@ -372,7 +328,10 @@ def _send_with_retry(
             )
 
     def _can_retry_on_another_token(self, request: requests.PreparedRequest) -> bool:
-        """Whether the authenticator can serve this request from a different credential now."""
+        """Whether the authenticator can serve this request from a different credential now.
+
+        Opted into by implementing `requests_native_auth.TokenRotatingAuthenticator`.
+        """
         authenticator = getattr(self._session, "auth", None)
         has_alternative_token = getattr(authenticator, "has_alternative_token", None)
         if not callable(has_alternative_token):
@@ -395,7 +354,7 @@ def _update_authenticator_from_response(
         has no way to learn that the server disagrees with its local bookkeeping. This is the
         feedback channel, mirroring what `LimiterMixin.send` does for the API budget.
 
-        Any authenticator implementing `ResponseAwareAuthenticator` opts in.
+        Opted into by implementing `requests_native_auth.ResponseAwareAuthenticator`.
         """
         authenticator = getattr(self._session, "auth", None)
         update_from_response = getattr(authenticator, "update_from_response", None)
diff --git a/airbyte_cdk/sources/streams/http/requests_native_auth/__init__.py b/airbyte_cdk/sources/streams/http/requests_native_auth/__init__.py
index 307f91f40c..5df3a0d058 100644
--- a/airbyte_cdk/sources/streams/http/requests_native_auth/__init__.py
+++ b/airbyte_cdk/sources/streams/http/requests_native_auth/__init__.py
@@ -3,6 +3,7 @@
 #
 
 from .oauth import Oauth2Authenticator, SingleUseRefreshTokenOauth2Authenticator
+from .protocols import ResponseAwareAuthenticator, TokenRotatingAuthenticator
 from .token import BasicHttpAuthenticator, MultipleTokenAuthenticator, TokenAuthenticator
 
 __all__ = [
@@ -11,4 +12,6 @@
     "TokenAuthenticator",
     "MultipleTokenAuthenticator",
     "BasicHttpAuthenticator",
+    "ResponseAwareAuthenticator",
+    "TokenRotatingAuthenticator",
 ]
diff --git a/airbyte_cdk/sources/streams/http/requests_native_auth/protocols.py b/airbyte_cdk/sources/streams/http/requests_native_auth/protocols.py
new file mode 100644
index 0000000000..a5107739d6
--- /dev/null
+++ b/airbyte_cdk/sources/streams/http/requests_native_auth/protocols.py
@@ -0,0 +1,55 @@
+#
+# Copyright (c) 2026 Airbyte, Inc., all rights reserved.
+#
+
+"""Optional capabilities an authenticator can offer the HTTP client.
+
+Authenticators are normally write-only from the client's point of view: they sign a request and
+that is the end of the conversation. An authenticator managing a quota, or several
+interchangeable credentials, needs more than that -- it has to learn what the server said, and
+it knows things about credential availability that the retry logic cannot work out on its own.
+
+These protocols are how it says so. Both are optional and dispatched structurally: `HttpClient`
+looks for the method and skips authenticators that do not define it, so implementing one does
+not require inheriting from anything or importing the client.
+"""
+
+from typing import Protocol, runtime_checkable
+
+import requests
+
+
+@runtime_checkable
+class ResponseAwareAuthenticator(Protocol):
+    """An authenticator that wants to see responses, not just sign requests.
+
+    Authenticators tracking per-token quota need a feedback channel: without one they can only
+    guess at the server's view of the quota and cannot tell that a token has been rejected.
+    `HttpClient` calls `update_from_response` once per attempt on any authenticator that
+    implements this, skipping replayed cache hits -- a cached response carries stale rate-limit
+    headers and consumed no quota.
+
+    Implementations must not raise for a response they cannot interpret, and must be safe to
+    call from multiple threads.
+    """
+
+    def update_from_response(
+        self, request: requests.PreparedRequest, response: requests.Response
+    ) -> None:
+        pass
+
+
+@runtime_checkable
+class TokenRotatingAuthenticator(Protocol):
+    """An authenticator holding several interchangeable credentials.
+
+    A rate-limit backoff is computed from the response alone, so it assumes the only way
+    forward is for that quota to come back. An authenticator with a spare credential knows
+    better. `HttpClient` asks before sleeping out a rate-limit window.
+
+    Implementations must only answer True when retrying immediately would actually use a
+    different credential -- otherwise the retry hammers the same rejected one.
+    """
+
+    def has_alternative_token(self, request: requests.PreparedRequest) -> bool:
+        pass

From 5637002fd1bb44497334577136d098fe53e1eb21 Mon Sep 17 00:00:00 2001
From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com>
Date: Mon, 17 Aug 2026 13:14:12 +0300
Subject: [PATCH 07/14] chore(low-code): drop an import left unused by the
 reverted restore tests

`ab_datetime_now` was only used by the two tests removed in 15d99fb7. The repo's ruff
config selects `I` and `F821` but not `F401`, so local lint stayed green and the
code-quality bot caught it instead.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../sources/declarative/auth/test_rate_limited_multiple_token.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py b/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py
index 48fd6901a4..ca851ce317 100644
--- a/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py
+++ b/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py
@@ -26,7 +26,6 @@
 )
 from airbyte_cdk.sources.streams.call_rate import HttpRequestRegexMatcher
 from airbyte_cdk.utils import AirbyteTracedException
-from airbyte_cdk.utils.datetime_helpers import ab_datetime_now
 
 QUOTA_STATUS_URL = "https://api.example.com/rate_limit"
 

From 1f9a9ed953ef8f30259122b181e3939a8dc70b2d Mon Sep 17 00:00:00 2001
From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com>
Date: Mon, 17 Aug 2026 13:19:46 +0300
Subject: [PATCH 08/14] fix(low-code): accept a limit header on the same terms
 as the reset it arrived with

`limit_header` was applied unconditionally, before the window comparison had a chance
to reject a stale response. A delayed response from an older window could therefore
overwrite the current limit, which feeds the proactive throttling reserve and the
capacity a later reset-only response refills the pool to.

This is the inconsistency I claimed had dissolved when the window rule went two-armed
in 51db59d4 -- it had, for `remaining`, but `limit` was still being taken from
responses whose reset value was being discarded. Now all three fields follow one rule:
a response only moves the window forward, never backward, and its numbers are accepted
only for the window it belongs to.

The assignment stays ahead of the rollover branch, which reads `state.limit` when a
fresh window arrives without a remaining header.

Testing: 926 tests pass across auth and streams.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../auth/rate_limited_multiple_token.py       |  7 +++++-
 .../auth/test_rate_limited_multiple_token.py  | 25 +++++++++++++++++++
 2 files changed, 31 insertions(+), 1 deletion(-)

diff --git a/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py b/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py
index 5b1e656675..86e33a0555 100644
--- a/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py
+++ b/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py
@@ -379,7 +379,12 @@ def update_from_response(
             state = self._states.get(token, {}).get(quota.name)
             if state is None:
                 return  # not seeded yet; the initial seeding is the more authoritative source
-            if limit is not None and limit > 0:
+            if limit is not None and limit > 0 and (reset_at is None or reset_at >= state.reset_at):
+                # A response from an older window carries that window's limit. Taking it would
+                # skew the throttling reserve and, on a later reset-only response, refill the
+                # pool to the wrong capacity -- so the limit is accepted on the same terms as
+                # the reset itself. Assigned before the rollover branch below, which reads
+                # `state.limit` when a fresh window arrives with no remaining header.
                 state.limit = limit
             if reset_at is not None and reset_at > state.reset_at:
                 # The quota window rolled over, so the local count is meaningless -- take the
diff --git a/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py b/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py
index ca851ce317..1d956252e9 100644
--- a/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py
+++ b/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py
@@ -438,6 +438,31 @@ def test_earlier_reset_in_response_never_moves_the_window_backwards(requests_moc
     assert int(state.reset_at.timestamp()) == current_window
 
 
+def test_limit_from_an_older_window_is_not_applied(requests_mock):
+    """The limit is accepted on the same terms as the reset it arrived with.
+
+    Taking a previous window's limit would skew the throttling reserve and, on a later
+    reset-only response, refill the pool to the wrong capacity.
+    """
+    requests_mock.get(QUOTA_STATUS_URL, json=_quota_status_body())
+    authenticator = _response_aware_authenticator(tokens=("token_1",))
+    request = authenticator(_prepared_request())
+    state = authenticator._states["token_1"]["rest"]
+    current_window = int(state.reset_at.timestamp())
+
+    authenticator.update_from_response(
+        request,
+        _response(
+            headers={
+                "X-RateLimit-Limit": "1000",
+                "X-RateLimit-Reset": str(current_window - 3600),
+            }
+        ),
+    )
+
+    assert state.limit == 5000
+
+
 def test_limit_header_updates_the_throttling_reserve(requests_mock):
     requests_mock.get(QUOTA_STATUS_URL, json=_quota_status_body())
     authenticator = _response_aware_authenticator(tokens=("token_1",))

From fb69c7e9a3218e4428c55d491f1ecbe7f0afbb7f Mon Sep 17 00:00:00 2001
From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com>
Date: Mon, 17 Aug 2026 13:38:39 +0300
Subject: [PATCH 09/14] fix(low-code): ignore a count from a window that has
 already rolled over

Round-2 review finding, and a consequence of the previous commit. Gating `limit` on the
window it arrived with left `remaining` as the one field that still accepted values from
any window, contradicting that commit's own claim that all three follow one rule.

The unconditional clamp was introduced for a real reason -- a 429 whose reset header
trails the value we hold must still zero the pool, or rotation silently stops working --
but it also accepts a *positive* count from a window that no longer exists. Since `min`
only ever tightens, that pins a freshly refilled pool to a dead window's count until the
next rollover an hour later. Measured before the fix: a pool that had just rolled over to
5000 was clamped to 3 by one late response from the previous window.

Reachable through exactly the in-flight-across-the-boundary race the rest of this
component is written around, and with 25 concurrent workers on hourly windows a response
straddling a rollover is close to certain once per window.

The clamp now requires the response to belong to the window we hold, with two exceptions
that preserve why it was unconditional: `remaining <= 0` is an exhaustion signal and is
never dropped, and a reset within RESET_SKEW_TOLERANCE (60s) still counts as the current
window, covering ordinary disagreement between the quota endpoint and response headers.

`test_earlier_reset_in_response_never_moves_the_window_backwards` pinned the behaviour
under discussion, so it is split: one test for a within-tolerance reset still clamping,
one for a rolled-over window being ignored. Verified the two properties that motivated
the unconditional clamp still hold -- exhaustion with an earlier reset still zeroes the
pool, and the end-to-end rotation still skips the wait.

Also adds the isinstance coverage the review asked for, so the `runtime_checkable`
protocols are exercised rather than merely declared.

Testing: 928 tests pass across auth and streams; ruff, format and mypy clean.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../auth/rate_limited_multiple_token.py       | 27 ++++++---
 .../auth/test_rate_limited_multiple_token.py  | 59 +++++++++++++++++--
 2 files changed, 74 insertions(+), 12 deletions(-)

diff --git a/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py b/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py
index 86e33a0555..0cffff3fb6 100644
--- a/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py
+++ b/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py
@@ -92,6 +92,10 @@ class RateLimitedMultipleTokenAuthenticator(DeclarativeAuthenticator):
     HEARTBEAT_INTERVAL = 60.0  # Log every 60s during exhaustion wait
     MAX_BUDGET_DELAY = 10.0  # Cap for the per-request proactive throttling delay
     MIN_EXHAUSTION_WAIT = 5.0  # Floor for the exhaustion wait, so stale reset timestamps can't cause a refresh busy-loop
+    # How far behind the reset we hold a response's reset may be while still counting as the
+    # current window. Covers ordinary disagreement between the quota endpoint and response
+    # headers; anything older is treated as belonging to a window that has already rolled over.
+    RESET_SKEW_TOLERANCE = timedelta(seconds=60)
 
     def __init__(
         self,
@@ -392,13 +396,22 @@ def update_from_response(
                 # is worth a full limit.
                 state.reset_at = reset_at
                 state.remaining = remaining if remaining is not None else state.limit
-            elif remaining is not None:
-                # Same window, or a response whose reset is older than what we hold. Only ever
-                # tighten the estimate: responses arrive out of order and a slow one carries a
-                # stale, higher count, so handing those calls back would let concurrent requests
-                # overspend. Clamping unconditionally also means an exhaustion signal is never
-                # dropped -- an earlier reset value must not be able to mask `remaining: 0`.
-                # The window itself is never moved backwards.
+            elif remaining is not None and (
+                remaining <= 0
+                or reset_at is None
+                or reset_at >= state.reset_at - self.RESET_SKEW_TOLERANCE
+            ):
+                # Same window: only ever tighten the estimate. Responses arrive out of order and
+                # a slow one carries a stale, higher count, so handing those calls back would let
+                # concurrent requests overspend. The window itself is never moved backwards.
+                #
+                # A count from a window that has already rolled over describes a window that no
+                # longer exists, and `min` would pin the fresh pool to it for the rest of the
+                # hour, so those are ignored -- with two exceptions. `remaining <= 0` is an
+                # exhaustion signal and must never be dropped, or a rate limit whose reset header
+                # trails the value we hold would silently stop rotation. And a reset within
+                # `RESET_SKEW_TOLERANCE` is treated as the current window, since the quota
+                # endpoint and the response headers can disagree by a little.
                 state.remaining = min(state.remaining, remaining)
 
     def has_alternative_token(self, request: requests.PreparedRequest) -> bool:
diff --git a/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py b/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py
index 1d956252e9..e3117a6c4e 100644
--- a/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py
+++ b/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py
@@ -25,6 +25,10 @@
     ModelToComponentFactory,
 )
 from airbyte_cdk.sources.streams.call_rate import HttpRequestRegexMatcher
+from airbyte_cdk.sources.streams.http.requests_native_auth import (
+    ResponseAwareAuthenticator,
+    TokenRotatingAuthenticator,
+)
 from airbyte_cdk.utils import AirbyteTracedException
 
 QUOTA_STATUS_URL = "https://api.example.com/rate_limit"
@@ -417,9 +421,9 @@ def test_later_reset_in_response_starts_a_new_window(requests_mock):
     assert int(state.reset_at.timestamp()) == next_window
 
 
-def test_earlier_reset_in_response_never_moves_the_window_backwards(requests_mock):
-    """A response carrying an older reset still clamps the count -- dropping it outright would
-    let a stale reset value mask an exhaustion signal -- but it must not rewind the window."""
+def test_slightly_earlier_reset_still_counts_as_the_current_window(requests_mock):
+    """The quota endpoint and the response headers can disagree by a little, so a reset just
+    behind the one we hold is the same window and its count still clamps."""
     requests_mock.get(QUOTA_STATUS_URL, json=_quota_status_body())
     authenticator = _response_aware_authenticator(tokens=("token_1",))
     request = authenticator(_prepared_request())
@@ -430,12 +434,46 @@ def test_earlier_reset_in_response_never_moves_the_window_backwards(requests_moc
     authenticator.update_from_response(
         request,
         _response(
-            headers={"X-RateLimit-Remaining": "7", "X-RateLimit-Reset": str(current_window - 3600)}
+            headers={"X-RateLimit-Remaining": "7", "X-RateLimit-Reset": str(current_window - 5)}
         ),
     )
 
     assert state.remaining == 7
-    assert int(state.reset_at.timestamp()) == current_window
+    assert int(state.reset_at.timestamp()) == current_window  # never moved backwards
+
+
+def test_count_from_a_rolled_over_window_does_not_clamp_the_fresh_pool(requests_mock):
+    """A slow response that lands after the window turned describes a window that no longer
+    exists. `min` would pin the refilled pool to that dead count for the rest of the hour, and
+    nothing short of the next rollover would undo it."""
+    requests_mock.get(QUOTA_STATUS_URL, json=_quota_status_body())
+    authenticator = _response_aware_authenticator(tokens=("token_1",))
+    request = authenticator(_prepared_request())
+    state = authenticator._states["token_1"]["rest"]
+    previous_window = int(state.reset_at.timestamp())
+
+    # the window turns over
+    authenticator.update_from_response(
+        request,
+        _response(
+            headers={
+                "X-RateLimit-Remaining": "5000",
+                "X-RateLimit-Reset": str(previous_window + 3600),
+            }
+        ),
+    )
+    assert state.remaining == 5000
+
+    # ...and only now does an in-flight response from the old window arrive
+    authenticator.update_from_response(
+        request,
+        _response(
+            headers={"X-RateLimit-Remaining": "3", "X-RateLimit-Reset": str(previous_window)}
+        ),
+    )
+
+    assert state.remaining == 5000
+    assert int(state.reset_at.timestamp()) == previous_window + 3600
 
 
 def test_limit_from_an_older_window_is_not_applied(requests_mock):
@@ -594,6 +632,17 @@ def test_graphql_response_updates_only_the_graphql_pool(requests_mock):
     assert authenticator._states["token_1"]["rest"].remaining == 5000
 
 
+def test_satisfies_both_authenticator_capability_protocols(requests_mock):
+    """`HttpClient` dispatches structurally, but the protocols are the declared contract -- an
+    implementation drifting from either one should fail here rather than silently stop being
+    consulted at runtime."""
+    requests_mock.get(QUOTA_STATUS_URL, json=_quota_status_body())
+    authenticator = _response_aware_authenticator()
+
+    assert isinstance(authenticator, ResponseAwareAuthenticator)
+    assert isinstance(authenticator, TokenRotatingAuthenticator)
+
+
 def test_has_alternative_token_when_sender_is_spent_and_another_is_not(requests_mock):
     requests_mock.get(QUOTA_STATUS_URL, json=_quota_status_body())
     authenticator = _response_aware_authenticator()

From af5e820494e1766a992b62202f8d79497836a831 Mon Sep 17 00:00:00 2001
From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com>
Date: Mon, 17 Aug 2026 14:08:06 +0300
Subject: [PATCH 10/14] fix(low-code): only honour a stale zero when the
 response was rate-limited

Round-3 review finding, and the last residual of the F9 fix. That fix kept one
unconditional path -- `remaining <= 0` was honoured whatever window it came from -- so a
rate limit whose reset header trails the value we hold could not silently stop rotation.
The property is right but the exception was wider than the case: a zero on a *successful*
response is just the last call of a window, and honouring it from a window that has
already rolled over parks a pool that has already refilled. Measured before the fix: a
200 carrying `X-RateLimit-Remaining: 0` and a dead window's reset took a refilled pool
from 5000 to 0 -- the same bug F9 fixed, leaking through the escape hatch.

The exception now also requires the response's status to be one the pool counts as
rate-limited, which is the configured expression of "this was a rejection, not a
successful call". After the fix the same stale zero from a 200 leaves the pool at 5000,
while from a 429 it still zeroes and rotates.

Consequence worth knowing: a pool that sets `remaining_header` but leaves
`exhaustion_status_codes` empty no longer gets the trailing-reset guarantee, since nothing
tells it which statuses mean rate-limited. Listing the API's unambiguous rate-limit status
is what buys that, which is what the field is for.

Testing: 929 tests pass across auth and streams; the three properties this arm has to
balance are each pinned by a test -- exhaustion with a trailing reset still lands, a
positive count from a dead window is ignored, and a stale zero from a 200 is ignored while
the same zero from a 429 is not.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../auth/rate_limited_multiple_token.py       | 17 ++++++----
 .../auth/test_rate_limited_multiple_token.py  | 33 +++++++++++++++++++
 2 files changed, 44 insertions(+), 6 deletions(-)

diff --git a/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py b/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py
index 0cffff3fb6..f56797109a 100644
--- a/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py
+++ b/airbyte_cdk/sources/declarative/auth/rate_limited_multiple_token.py
@@ -397,7 +397,7 @@ def update_from_response(
                 state.reset_at = reset_at
                 state.remaining = remaining if remaining is not None else state.limit
             elif remaining is not None and (
-                remaining <= 0
+                (remaining <= 0 and response.status_code in quota.exhaustion_status_codes)
                 or reset_at is None
                 or reset_at >= state.reset_at - self.RESET_SKEW_TOLERANCE
             ):
@@ -407,11 +407,16 @@ def update_from_response(
                 #
                 # A count from a window that has already rolled over describes a window that no
                 # longer exists, and `min` would pin the fresh pool to it for the rest of the
-                # hour, so those are ignored -- with two exceptions. `remaining <= 0` is an
-                # exhaustion signal and must never be dropped, or a rate limit whose reset header
-                # trails the value we hold would silently stop rotation. And a reset within
-                # `RESET_SKEW_TOLERANCE` is treated as the current window, since the quota
-                # endpoint and the response headers can disagree by a little.
+                # hour, so those are ignored -- with two exceptions.
+                #
+                # First, a zero on a response the pool counts as rate-limited is an exhaustion
+                # signal and must never be dropped, or a rate limit whose reset header trails the
+                # value we hold would silently stop rotation. The status check is what keeps that
+                # narrow: a zero on a *successful* response is just the last call of a window, and
+                # honouring it from a dead window would park a pool that has already refilled.
+                #
+                # Second, a reset within `RESET_SKEW_TOLERANCE` is treated as the current window,
+                # since the quota endpoint and the response headers can disagree by a little.
                 state.remaining = min(state.remaining, remaining)
 
     def has_alternative_token(self, request: requests.PreparedRequest) -> bool:
diff --git a/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py b/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py
index e3117a6c4e..0a45800918 100644
--- a/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py
+++ b/unit_tests/sources/declarative/auth/test_rate_limited_multiple_token.py
@@ -442,6 +442,39 @@ def test_slightly_earlier_reset_still_counts_as_the_current_window(requests_mock
     assert int(state.reset_at.timestamp()) == current_window  # never moved backwards
 
 
+def test_stale_zero_from_a_successful_response_does_not_park_a_refilled_pool(requests_mock):
+    """A zero on a 200 is just the last call of a window, not a rate-limit rejection.
+
+    The exhaustion exception to the window check exists so a rate limit whose reset header
+    trails cannot stop rotation; honouring a zero from *any* response would let a dead
+    window's final count park a pool that has already refilled -- the F9 bug via the escape
+    hatch. The status is what tells the two apart.
+    """
+    requests_mock.get(QUOTA_STATUS_URL, json=_quota_status_body())
+    authenticator = _response_aware_authenticator(tokens=("token_1",))
+    request = authenticator(_prepared_request())
+    state = authenticator._states["token_1"]["rest"]
+    previous_window = int(state.reset_at.timestamp())
+    authenticator.update_from_response(
+        request,
+        _response(
+            headers={
+                "X-RateLimit-Remaining": "5000",
+                "X-RateLimit-Reset": str(previous_window + 3600),
+            }
+        ),
+    )
+    assert state.remaining == 5000
+
+    stale = {"X-RateLimit-Remaining": "0", "X-RateLimit-Reset": str(previous_window)}
+    authenticator.update_from_response(request, _response(status_code=200, headers=stale))
+    assert state.remaining == 5000, "a zero from a successful dead-window response must be ignored"
+
+    # ...but the same zero on a rate-limited response is an exhaustion signal and still lands
+    authenticator.update_from_response(request, _response(status_code=429, headers=stale))
+    assert state.remaining == 0
+
+
 def test_count_from_a_rolled_over_window_does_not_clamp_the_fresh_pool(requests_mock):
     """A slow response that lands after the window turned describes a window that no longer
     exists. `min` would pin the refilled pool to that dead count for the rest of the hour, and

From 36dc700edf79cefc7d7b7adb046b6f23c3c04a84 Mon Sep 17 00:00:00 2001
From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com>
Date: Mon, 17 Aug 2026 14:44:28 +0300
Subject: [PATCH 11/14] docs(low-code): describe both effects of
 exhaustion_status_codes

Round-4 review finding. The description covered only the original effect -- setting the
pool to zero when a rate-limited response carries no remaining count. Since af5e8204 the
field also decides which responses may report a zero for a window that has already
elapsed, which is what keeps a rate limit whose reset header trails from silently
stopping rotation, and it is the reason leaving the list empty gives up that guarantee.

The schema is the only place a manifest author encounters this, so it is the one place
the two effects have to be stated. Model description hand-edited to match the YAML
verbatim, as with the rest of this PR's generated-file changes.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../sources/declarative/declarative_component_schema.yaml       | 2 +-
 .../sources/declarative/models/declarative_component_schema.py  | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml
index 7de66bcf7f..091c694d23 100644
--- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml
+++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml
@@ -517,7 +517,7 @@ definitions:
           - "X-RateLimit-Limit"
       exhaustion_status_codes:
         title: Exhaustion Status Codes
-        description: Response status codes that mean this token's pool is spent, applied when the response carries no remaining header. The pool is set to zero so the next request rotates to another token instead of waiting out the reset window. Only list codes the API uses exclusively for rate limiting -- a code that also signals other failures would park a healthy token.
+        description: Response status codes that mean this token's pool is spent. These have two effects. A response carrying one of them but no remaining count sets the pool to zero, so the next request rotates to another token instead of waiting out the reset window. They also mark which responses may report a zero for a quota window that has already elapsed, so a rate limit whose reset header trails the value being held still stops the token being used; a zero on any other response is treated as the last call of a finished window and ignored. Leaving this empty means such trailing rejections are ignored unless their reset is within the skew tolerance of the current window. Only list codes the API uses exclusively for rate limiting -- a code that also signals other failures would park a healthy token.
         type: array
         items:
           type: integer
diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py
index 69ad3dcfe9..37eb50b7b1 100644
--- a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py
+++ b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py
@@ -629,7 +629,7 @@ class TokenQuota(BaseModel):
     )
     exhaustion_status_codes: Optional[List[int]] = Field(
         None,
-        description="Response status codes that mean this token's pool is spent, applied when the response carries no remaining header. The pool is set to zero so the next request rotates to another token instead of waiting out the reset window. Only list codes the API uses exclusively for rate limiting -- a code that also signals other failures would park a healthy token.",
+        description="Response status codes that mean this token's pool is spent. These have two effects. A response carrying one of them but no remaining count sets the pool to zero, so the next request rotates to another token instead of waiting out the reset window. They also mark which responses may report a zero for a quota window that has already elapsed, so a rate limit whose reset header trails the value being held still stops the token being used; a zero on any other response is treated as the last call of a finished window and ignored. Leaving this empty means such trailing rejections are ignored unless their reset is within the skew tolerance of the current window. Only list codes the API uses exclusively for rate limiting -- a code that also signals other failures would park a healthy token.",
         examples=[[429]],
         title="Exhaustion Status Codes",
     )

From 7243bfd91befd174da7f874632a6c972c9ef2afd Mon Sep 17 00:00:00 2001
From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com>
Date: Mon, 17 Aug 2026 16:11:08 +0300
Subject: [PATCH 12/14] refactor(low-code): dispatch on the authenticator
 protocols instead of getattr

The two protocols were referenced only from docstrings, so the contract they declared and
the contract actually enforced were separate things: renaming a method on a protocol
changed nothing, because the string inside `getattr` was the real interface. Making the
dispatch sites check the protocols collapses those into one object.

`protocols` imports nothing from the CDK -- only `typing` and `requests` -- so importing
it here cannot cycle whatever else lands in `requests_native_auth`, including an
authenticator that needs `HttpClient` the way the declarative one does. That is why the
import targets the leaf module rather than the package.

Behaviour is unchanged. `runtime_checkable` protocols check method presence via attribute
lookup, so all three cases resolve exactly as the `getattr` form did: a plain
authenticator is skipped, one whose class defines the method is dispatched, and so is one
with the method assigned on the instance. Verified, along with the end-to-end rate-limit
rotation and the stale-zero handling.

Worth stating the limit: this checks presence, not signature, so an implementation with
the wrong arity still passes -- the same guarantee `getattr` gave, just expressed once.

Testing: 929 tests pass across auth and streams; ruff, format and mypy clean.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../sources/streams/http/http_client.py       | 22 ++++++++++++-------
 1 file changed, 14 insertions(+), 8 deletions(-)

diff --git a/airbyte_cdk/sources/streams/http/http_client.py b/airbyte_cdk/sources/streams/http/http_client.py
index c23dfcb9e5..263a4ea1ba 100644
--- a/airbyte_cdk/sources/streams/http/http_client.py
+++ b/airbyte_cdk/sources/streams/http/http_client.py
@@ -50,6 +50,14 @@
     rate_limit_default_backoff_handler,
     user_defined_backoff_handler,
 )
+
+# Imported from the leaf module rather than the package: `protocols` pulls in nothing from the
+# CDK, so this import cannot cycle no matter what else lands in `requests_native_auth` -- an
+# authenticator there needing `HttpClient` (as the declarative one does) stays safe.
+from airbyte_cdk.sources.streams.http.requests_native_auth.protocols import (
+    ResponseAwareAuthenticator,
+    TokenRotatingAuthenticator,
+)
 from airbyte_cdk.sources.utils.types import JsonType
 from airbyte_cdk.utils.airbyte_secrets_utils import filter_secrets
 from airbyte_cdk.utils.constants import ENV_REQUEST_CACHE_PATH
@@ -330,14 +338,13 @@ def _send_with_retry(
     def _can_retry_on_another_token(self, request: requests.PreparedRequest) -> bool:
         """Whether the authenticator can serve this request from a different credential now.
 
-        Opted into by implementing `requests_native_auth.TokenRotatingAuthenticator`.
+        Opted into by implementing `TokenRotatingAuthenticator`.
         """
         authenticator = getattr(self._session, "auth", None)
-        has_alternative_token = getattr(authenticator, "has_alternative_token", None)
-        if not callable(has_alternative_token):
+        if not isinstance(authenticator, TokenRotatingAuthenticator):
             return False
         try:
-            return bool(has_alternative_token(request))
+            return bool(authenticator.has_alternative_token(request))
         except Exception:
             # Falling back to the computed wait is always safe, so never fail a retry over this.
             self._logger.debug(
@@ -354,18 +361,17 @@ def _update_authenticator_from_response(
         has no way to learn that the server disagrees with its local bookkeeping. This is the
         feedback channel, mirroring what `LimiterMixin.send` does for the API budget.
 
-        Opted into by implementing `requests_native_auth.ResponseAwareAuthenticator`.
+        Opted into by implementing `ResponseAwareAuthenticator`.
         """
         authenticator = getattr(self._session, "auth", None)
-        update_from_response = getattr(authenticator, "update_from_response", None)
-        if not callable(update_from_response):
+        if not isinstance(authenticator, ResponseAwareAuthenticator):
             return
         if getattr(response, "from_cache", False):
             # A replayed cached response carries the rate-limit headers from whenever it was
             # first fetched and consumed no quota of its own.
             return
         try:
-            update_from_response(request, response)
+            authenticator.update_from_response(request, response)
         except Exception:
             # Quota bookkeeping must never turn an otherwise fine response into a failure. Warn
             # once so a persistently broken update -- which silently degrades the connector back

From ef41e784f77f91a9e27cd7603b20e69bcbb36525 Mon Sep 17 00:00:00 2001
From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com>
Date: Mon, 17 Aug 2026 16:41:48 +0300
Subject: [PATCH 13/14] docs(low-code): state how the authenticator capability
 protocols are matched

The module described dispatch as "looks for the method", which was written when the
dispatch sites used `getattr`. Since 7243bfd9 they check the protocol with `isinstance`,
and from Python 3.12 that resolves protocol members through `inspect.getattr_static`,
which does not run `__getattr__`. A delegating authenticator that only exposes the
method dynamically therefore satisfies the protocol on 3.10/3.11 and is silently skipped
on 3.12/3.13 -- measured across all four interpreters in the CI matrix. Connector authors
read this module before writing an authenticator, so the constraint belongs here, next to
the note that presence rather than signature is what gets checked.

Also corrects the blanket "behaviour is unchanged" in 7243bfd9: the three cases that
commit enumerates do resolve identically, the general claim does not.

Docstring only; no behaviour change. 216 tests pass across auth and http_client; ruff,
format and mypy clean.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../streams/http/requests_native_auth/protocols.py    | 11 +++++++++--
 1 file changed, 9 insertions(+), 2 deletions(-)

diff --git a/airbyte_cdk/sources/streams/http/requests_native_auth/protocols.py b/airbyte_cdk/sources/streams/http/requests_native_auth/protocols.py
index a5107739d6..7507b370b2 100644
--- a/airbyte_cdk/sources/streams/http/requests_native_auth/protocols.py
+++ b/airbyte_cdk/sources/streams/http/requests_native_auth/protocols.py
@@ -10,8 +10,15 @@
 it knows things about credential availability that the retry logic cannot work out on its own.
 
 These protocols are how it says so. Both are optional and dispatched structurally: `HttpClient`
-looks for the method and skips authenticators that do not define it, so implementing one does
-not require inheriting from anything or importing the client.
+checks the authenticator against the protocol and skips one that does not satisfy it, so
+implementing one does not require inheriting from anything or importing the client.
+
+Define the methods on the class or set them on the instance. Dispatch is an `isinstance` check,
+and from Python 3.12 those resolve protocol members with `inspect.getattr_static`, which does
+not run `__getattr__` -- so a delegating authenticator that only exposes the method dynamically
+satisfies the protocol on 3.10/3.11 and is silently skipped on 3.12+. Presence is all that is
+checked either way: an implementation with the wrong signature still dispatches, and then fails
+at the call.
 """
 
 from typing import Protocol, runtime_checkable

From 3c6e88200b96c1a76b75414a05360e14113996e4 Mon Sep 17 00:00:00 2001
From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com>
Date: Tue, 18 Aug 2026 13:20:36 +0300
Subject: [PATCH 14/14] chore(ci): retrigger checks after a transient
 Dockerfile-template 404

`Pytest (All)` failed on all four interpreters at ef41e784 with

  ConnectorImageBuildError: Could not locate local dockerfile templates and failed to
  download Dockerfile templates from github: 404 Client Error: Not Found for url:
  https://github.com/airbytehq/airbyte/raw/master/docker-images//Dockerfile.manifest-only-connector

The two failures are `test_docker_image_build_and_spec` and
`test_docker_image_build_and_check` in the `source_pokeapi_w_components_py` resource, which
fetch that template at run time. The URL serves 200 now, the same tests passed on this
branch 29 minutes earlier at 7243bfd9, and the only change since was a docstring -- so the
404 was upstream and transient, not a regression.

Empty commit; no content change.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>