From 0fa0ace61871ad6a37395159e68c8da0d7313392 Mon Sep 17 00:00:00 2001
From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com>
Date: Wed, 19 Aug 2026 16:29:21 +0300
Subject: [PATCH 1/5] feat(low-code): cap the wait WaitUntilTimeFromHeader is
 willing to return

A rate-limit backoff derived from a reset header is unbounded. WaitTimeFromHeader
has had max_waiting_time_in_seconds for this since it was written -- raise rather
than sleep past a limit the connector is willing to accept -- but the sibling
strategy that reads an absolute reset timestamp has no equivalent, so a response
carrying a reset an hour out sleeps for an hour with nothing able to stop it.

Nothing else can bound it either. DefaultErrorHandler.max_time cannot: the sleep
happens inside user_defined_backoff_handler's on_backoff callback while the
backoff library's own interval is 0, so the budget check never sees it. An
authenticator's own wait bound cannot: that governs the proactive path, where
local counters say the quota is spent before a request goes out, and this is the
reactive path, where the server rejected a request the counters thought was fine.

Both fields are now interpolatable, which is the point of the change. One
manifest can then give a connection check a tighter bound than a sync -- a check
is interactive and should fail fast with an actionable message, a sync can afford
to sleep through a rate-limit window rather than fail -- by overriding a single
config value for the duration of the check.

Two details worth knowing when reading the diff. The cap compares against the
wait the strategy is about to return rather than the raw header, because unlike
Retry-After the header here is an absolute timestamp and only the difference is a
duration; it is also applied after the min_wait floor, so a cap below the floor
still wins. And the guard is `is not None` rather than a truthiness check, since
0 is the value a caller uses to say "never wait" -- WaitTimeFromHeader silently
ignored it, which is fixed here too.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../declarative_component_schema.yaml         | 20 +++-
 .../models/declarative_component_schema.py    | 12 ++-
 .../parsers/model_to_component_factory.py     |  1 +
 .../wait_time_from_header_backoff_strategy.py | 29 +++++-
 ...until_time_from_header_backoff_strategy.py | 49 +++++++++-
 .../test_wait_until_time_from_header.py       | 92 +++++++++++++++++++
 6 files changed, 190 insertions(+), 13 deletions(-)

diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml
index 091c694d23..05f89096da 100644
--- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml
+++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml
@@ -4641,10 +4641,15 @@ definitions:
           - "([-+]?\\d+)"
       max_waiting_time_in_seconds:
         title: Max Waiting Time in Seconds
-        description: Given the value extracted from the header is greater than this value, stop the stream.
-        type: number
+        description: Stop the stream instead of waiting, when the value extracted from the header is greater than this value. Can be a hardcoded number or a string interpolated from the connector config, which lets an operation that must answer quickly, such as a connection check, use a tighter bound than a sync. A value of 0 means never wait.
+        anyOf:
+          - type: number
+          - type: string
+        interpolation_context:
+          - config
         examples:
           - 3600
+          - "{{ config['max_waiting_time'] * 60 }}"
       $parameters:
         type: object
         additionalProperties: true
@@ -4770,6 +4775,17 @@ definitions:
           - config
         examples:
           - "([-+]?\\d+)"
+      max_waiting_time_in_seconds:
+        title: Max Waiting Time in Seconds
+        description: Stop the stream instead of waiting, when the wait this strategy computes is longer than this value. The comparison is against the computed wait rather than the raw header, since the header holds an absolute timestamp, and it is applied after `min_wait`, so a cap below the floor still wins. Can be a hardcoded number or a string interpolated from the connector config, which lets an operation that must answer quickly, such as a connection check, use a tighter bound than a sync. A value of 0 means never wait.
+        anyOf:
+          - type: number
+          - type: string
+        interpolation_context:
+          - config
+        examples:
+          - 3600
+          - "{{ config['max_waiting_time'] * 60 }}"
       $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 37eb50b7b1..67888b241d 100644
--- a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py
+++ b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py
@@ -1434,10 +1434,10 @@ class WaitTimeFromHeader(BaseModel):
         examples=["([-+]?\\d+)"],
         title="Extraction Regex",
     )
-    max_waiting_time_in_seconds: Optional[float] = Field(
+    max_waiting_time_in_seconds: Optional[Union[float, str]] = Field(
         None,
-        description="Given the value extracted from the header is greater than this value, stop the stream.",
-        examples=[3600],
+        description="Stop the stream instead of waiting, when the value extracted from the header is greater than this value. Can be a hardcoded number or a string interpolated from the connector config, which lets an operation that must answer quickly, such as a connection check, use a tighter bound than a sync. A value of 0 means never wait.",
+        examples=[3600, "{{ config['max_waiting_time'] * 60 }}"],
         title="Max Waiting Time in Seconds",
     )
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
@@ -1463,6 +1463,12 @@ class WaitUntilTimeFromHeader(BaseModel):
         examples=["([-+]?\\d+)"],
         title="Extraction Regex",
     )
+    max_waiting_time_in_seconds: Optional[Union[float, str]] = Field(
+        None,
+        description="Stop the stream instead of waiting, when the wait this strategy computes is longer than this value. The comparison is against the computed wait rather than the raw header, since the header holds an absolute timestamp, and it is applied after `min_wait`, so a cap below the floor still wins. Can be a hardcoded number or a string interpolated from the connector config, which lets an operation that must answer quickly, such as a connection check, use a tighter bound than a sync. A value of 0 means never wait.",
+        examples=[3600, "{{ config['max_waiting_time'] * 60 }}"],
+        title="Max Waiting Time in Seconds",
+    )
     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 a96c406d8c..2512c34470 100644
--- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py
+++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py
@@ -4349,6 +4349,7 @@ def create_wait_until_time_from_header(
             config=config,
             min_wait=model.min_wait,
             regex=model.regex,
+            max_waiting_time_in_seconds=model.max_waiting_time_in_seconds,
         )
 
     def get_message_repository(self) -> MessageRepository:
diff --git a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_time_from_header_backoff_strategy.py b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_time_from_header_backoff_strategy.py
index 5cda96a4de..b8eb0cf5c0 100644
--- a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_time_from_header_backoff_strategy.py
+++ b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_time_from_header_backoff_strategy.py
@@ -35,13 +35,21 @@ class WaitTimeFromHeaderBackoffStrategy(BackoffStrategy):
     parameters: InitVar[Mapping[str, Any]]
     config: Config
     regex: Optional[Union[InterpolatedString, str]] = None
-    max_waiting_time_in_seconds: Optional[float] = None
+    max_waiting_time_in_seconds: Optional[Union[float, InterpolatedString, str]] = None
 
     def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         self.regex = (
             InterpolatedString.create(self.regex, parameters=parameters) if self.regex else None
         )
         self.header = InterpolatedString.create(self.header, parameters=parameters)
+        self._max_waiting_time_in_seconds = (
+            self.max_waiting_time_in_seconds
+            if self.max_waiting_time_in_seconds is None
+            or isinstance(self.max_waiting_time_in_seconds, InterpolatedString)
+            else InterpolatedString.create(
+                str(self.max_waiting_time_in_seconds), parameters=parameters
+            )
+        )
 
     def backoff_time(
         self,
@@ -57,14 +65,25 @@ def backoff_time(
         header_value = None
         if isinstance(response_or_exception, requests.Response):
             header_value = get_numeric_value_from_header(response_or_exception, header, regex)
+            max_waiting_time = self._eval_max_waiting_time()
+            # `is not None` rather than a truthiness check, so that 0 means "never wait" instead
+            # of silently disabling the cap.
             if (
-                self.max_waiting_time_in_seconds
-                and header_value
-                and header_value >= self.max_waiting_time_in_seconds
+                max_waiting_time is not None
+                and header_value is not None
+                and header_value >= max_waiting_time
             ):
                 raise AirbyteTracedException(
-                    internal_message=f"Rate limit wait time {header_value} is greater than max waiting time of {self.max_waiting_time_in_seconds} seconds. Stopping the stream...",
+                    internal_message=f"Rate limit wait time {header_value} is greater than max waiting time of {max_waiting_time} seconds. Stopping the stream...",
                     message="The rate limit is greater than max waiting time has been reached.",
                     failure_type=FailureType.transient_error,
                 )
         return header_value
+
+    def _eval_max_waiting_time(self) -> Optional[float]:
+        if self._max_waiting_time_in_seconds is None:
+            return None
+        evaluated = self._max_waiting_time_in_seconds.eval(self.config)
+        if evaluated is None or evaluated == "":
+            return None
+        return float(evaluated)
diff --git a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_until_time_from_header_backoff_strategy.py b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_until_time_from_header_backoff_strategy.py
index 1220e198f5..478269dd2a 100644
--- a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_until_time_from_header_backoff_strategy.py
+++ b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_until_time_from_header_backoff_strategy.py
@@ -10,6 +10,7 @@
 
 import requests
 
+from airbyte_cdk.models import FailureType
 from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString
 from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies.header_helper import (
     get_numeric_value_from_header,
@@ -18,6 +19,7 @@
     BackoffStrategy,
 )
 from airbyte_cdk.sources.types import Config
+from airbyte_cdk.utils import AirbyteTracedException
 
 
 @dataclass
@@ -30,6 +32,7 @@ class WaitUntilTimeFromHeaderBackoffStrategy(BackoffStrategy):
         header (str): header to read wait time from
         min_wait (Optional[float]): minimum time to wait for safety
         regex (Optional[str]): optional regex to apply on the header to extract its value
+        max_waiting_time_in_seconds (Optional[float]): stop the stream rather than wait longer than this
     """
 
     header: Union[InterpolatedString, str]
@@ -37,6 +40,7 @@ class WaitUntilTimeFromHeaderBackoffStrategy(BackoffStrategy):
     config: Config
     min_wait: Optional[Union[float, InterpolatedString, str]] = None
     regex: Optional[Union[InterpolatedString, str]] = None
+    max_waiting_time_in_seconds: Optional[Union[float, InterpolatedString, str]] = None
 
     def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         self.header = InterpolatedString.create(self.header, parameters=parameters)
@@ -45,6 +49,14 @@ def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         )
         if not isinstance(self.min_wait, InterpolatedString):
             self.min_wait = InterpolatedString.create(str(self.min_wait), parameters=parameters)
+        self._max_waiting_time_in_seconds = (
+            self.max_waiting_time_in_seconds
+            if self.max_waiting_time_in_seconds is None
+            or isinstance(self.max_waiting_time_in_seconds, InterpolatedString)
+            else InterpolatedString.create(
+                str(self.max_waiting_time_in_seconds), parameters=parameters
+            )
+        )
 
     def backoff_time(
         self,
@@ -63,15 +75,46 @@ def backoff_time(
             wait_until = get_numeric_value_from_header(response_or_exception, header, regex)
         min_wait = self.min_wait.eval(self.config)  # type: ignore # header is always cast to an interpolated string
         if wait_until is None or not wait_until:
-            return float(min_wait) if min_wait else None
+            return self._capped(float(min_wait)) if min_wait else None
         if (isinstance(wait_until, str) and wait_until.isnumeric()) or isinstance(
             wait_until, numbers.Number
         ):
             wait_time = float(wait_until) - now
         else:
-            return float(min_wait)
+            return self._capped(float(min_wait))
         if min_wait:
-            return float(max(wait_time, min_wait))
+            return self._capped(float(max(wait_time, min_wait)))
         elif wait_time < 0:
             return None
+        return self._capped(wait_time)
+
+    def _capped(self, wait_time: float) -> float:
+        """Raise rather than wait longer than `max_waiting_time_in_seconds`.
+
+        The cap is compared against the wait this strategy is about to return, not against the
+        raw header: unlike `Retry-After`, the header here is an absolute timestamp, so only the
+        computed difference is a duration. It is also applied after the `min_wait` floor, so a
+        cap below the floor wins -- a caller asking never to wait more than N seconds means it,
+        even when the floor would otherwise round the wait up past N.
+        """
+        max_waiting_time = self._eval_max_waiting_time()
+        if max_waiting_time is not None and wait_time > max_waiting_time:
+            raise AirbyteTracedException(
+                internal_message=(
+                    f"Rate limit wait time {wait_time}s is greater than the maximum of "
+                    f"{max_waiting_time}s this stream is allowed to wait. Stopping the stream..."
+                ),
+                message="The rate limit wait time is longer than the connector is allowed to wait.",
+                failure_type=FailureType.transient_error,
+            )
         return wait_time
+
+    def _eval_max_waiting_time(self) -> Optional[float]:
+        if self._max_waiting_time_in_seconds is None:
+            return None
+        evaluated = self._max_waiting_time_in_seconds.eval(self.config)
+        # `is None` rather than a truthiness check: 0 is a meaningful cap -- "never wait" -- and
+        # the equivalent field on WaitTimeFromHeader silently disables itself when set to 0.
+        if evaluated is None or evaluated == "":
+            return None
+        return float(evaluated)
diff --git a/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_until_time_from_header.py b/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_until_time_from_header.py
index 4c4c5a6f71..e6436ac88e 100644
--- a/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_until_time_from_header.py
+++ b/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_until_time_from_header.py
@@ -8,9 +8,11 @@
 import pytest
 import requests
 
+from airbyte_cdk.models import FailureType
 from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies.wait_until_time_from_header_backoff_strategy import (
     WaitUntilTimeFromHeaderBackoffStrategy,
 )
+from airbyte_cdk.utils import AirbyteTracedException
 
 SOME_BACKOFF_TIME = 60
 REGEX = "[-+]?\\d+"
@@ -122,3 +124,93 @@ def test_wait_untiltime_from_header(
     )
     backoff = backoff_strategy.backoff_time(response_mock, 1)
     assert backoff == expected_backoff_time
+
+
+NOW = 1600000000.0
+IN_60_SECONDS = NOW + 60
+
+
+def _response(wait_until=IN_60_SECONDS):
+    response = MagicMock(spec=requests.Response)
+    response.headers = {"wait_until": wait_until}
+    return response
+
+
+def _strategy(max_waiting_time_in_seconds, min_wait=None, config=None):
+    return WaitUntilTimeFromHeaderBackoffStrategy(
+        header="wait_until",
+        min_wait=min_wait,
+        parameters={},
+        config=config if config is not None else {},
+        max_waiting_time_in_seconds=max_waiting_time_in_seconds,
+    )
+
+
+@pytest.mark.parametrize(
+    "max_waiting_time_in_seconds, expected",
+    [
+        pytest.param(None, 60, id="no_cap_waits"),
+        pytest.param(3600, 60, id="cap_above_the_wait_waits"),
+        pytest.param(60, 60, id="cap_equal_to_the_wait_waits"),
+        pytest.param(30, "raises", id="cap_below_the_wait_raises"),
+        pytest.param(0, "raises", id="zero_cap_never_waits"),
+    ],
+)
+@patch("time.time", return_value=NOW)
+def test_max_waiting_time_in_seconds(time_mock, max_waiting_time_in_seconds, expected):
+    """The cap is what lets one operation refuse a wait that another operation would accept.
+
+    `0` has to raise rather than switch the cap off: it is the value a caller uses to say "never
+    wait", and the equivalent field on WaitTimeFromHeader read it as falsy and ignored it.
+    """
+    strategy = _strategy(max_waiting_time_in_seconds)
+
+    if expected == "raises":
+        with pytest.raises(AirbyteTracedException) as exc_info:
+            strategy.backoff_time(_response(), 1)
+        assert exc_info.value.failure_type == FailureType.transient_error
+    else:
+        assert strategy.backoff_time(_response(), 1) == expected
+
+
+@patch("time.time", return_value=NOW)
+def test_cap_is_applied_after_the_min_wait_floor(time_mock):
+    """`min_wait` can round a short wait up past the cap. The cap wins -- a caller that says it
+    will never wait longer than N seconds means it, floor or no floor."""
+    strategy = _strategy(max_waiting_time_in_seconds=30, min_wait=60)
+
+    with pytest.raises(AirbyteTracedException):
+        strategy.backoff_time(_response(NOW + 1), 1)
+
+
+@patch("time.time", return_value=NOW)
+def test_cap_applies_to_the_min_wait_fallback_when_the_header_is_absent(time_mock):
+    """With no usable header the strategy falls back to `min_wait`; that fallback is a wait like
+    any other and has to respect the cap."""
+    response = MagicMock(spec=requests.Response)
+    response.headers = {}
+    strategy = _strategy(max_waiting_time_in_seconds=30, min_wait=60)
+
+    with pytest.raises(AirbyteTracedException):
+        strategy.backoff_time(response, 1)
+
+
+@pytest.mark.parametrize(
+    "config, expected",
+    [
+        pytest.param({"max_waiting_time": 120}, 60, id="sync_budget_allows_the_wait"),
+        pytest.param({"max_waiting_time": 0}, "raises", id="check_budget_refuses_the_wait"),
+    ],
+)
+@patch("time.time", return_value=NOW)
+def test_cap_is_interpolated_from_config(time_mock, config, expected):
+    """The reason the field is interpolatable: one manifest, and an operation that has to answer
+    quickly -- a connection check overriding `max_waiting_time` to 0 -- refuses a wait the same
+    stream takes happily during a sync."""
+    strategy = _strategy("{{ config['max_waiting_time'] * 60 }}", config=config)
+
+    if expected == "raises":
+        with pytest.raises(AirbyteTracedException):
+            strategy.backoff_time(_response(), 1)
+    else:
+        assert strategy.backoff_time(_response(), 1) == expected

From e56d75afcc22a600d3445ea6ad1aeecf84f5231b Mon Sep 17 00:00:00 2001
From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com>
Date: Wed, 19 Aug 2026 17:03:46 +0300
Subject: [PATCH 2/5] fix(low-code): harden and de-duplicate the
 max_waiting_time_in_seconds cap

Review follow-ups on the WaitUntilTimeFromHeader cap:

- Raise a config_error when an interpolated cap cannot be evaluated. The cap
  is only read while handling an error that was already going to be retried,
  so a missing config key or a non-numeric value used to surface as an
  unhandled jinja UndefinedError or ValueError the first time an API rate
  limited a sync that had been running fine.
- Extract the shared evaluation into max_waiting_time_helper so the two
  strategies cannot drift, and give both the same user-facing message.
- Drop the check-vs-sync framing from the two field descriptions: nothing in
  a manifest can vary a config value per operation until CheckStream
  config_overrides lands, so the schema promised something authors cannot do.
  Say "greater than or equal to" for WaitTimeFromHeader, which is what its
  comparison has always done, and record why the two strategies differ at the
  boundary.
- Remove the unreachable string branch in WaitUntilTimeFromHeader.backoff_time:
  get_numeric_value_from_header returns a float or None, never a str.
- Cover the behaviour change on WaitTimeFromHeader (a cap of 0 now means
  "never wait") and its new interpolation, plus the config_error path on both
  strategies.
- Tidy the docstrings for the changed fields and drop a no-op ternary in
  create_wait_time_from_header.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../declarative_component_schema.yaml         |  4 +-
 .../models/declarative_component_schema.py    |  4 +-
 .../parsers/model_to_component_factory.py     |  4 +-
 .../max_waiting_time_helper.py                | 72 +++++++++++++++++++
 .../wait_time_from_header_backoff_strategy.py | 42 +++++------
 ...until_time_from_header_backoff_strategy.py | 41 ++++-------
 .../test_wait_time_from_header.py             | 67 +++++++++++++++++
 .../test_wait_until_time_from_header.py       | 23 ++++++
 8 files changed, 201 insertions(+), 56 deletions(-)
 create mode 100644 airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/max_waiting_time_helper.py

diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml
index 05f89096da..0a51d1e71e 100644
--- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml
+++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml
@@ -4641,7 +4641,7 @@ definitions:
           - "([-+]?\\d+)"
       max_waiting_time_in_seconds:
         title: Max Waiting Time in Seconds
-        description: Stop the stream instead of waiting, when the value extracted from the header is greater than this value. Can be a hardcoded number or a string interpolated from the connector config, which lets an operation that must answer quickly, such as a connection check, use a tighter bound than a sync. A value of 0 means never wait.
+        description: Stop the stream instead of waiting, when the value extracted from the header is greater than or equal to this value. Can be a hardcoded number, or a string interpolated from the connector config so that the bound can be changed without a connector release. A value of 0 means never wait.
         anyOf:
           - type: number
           - type: string
@@ -4777,7 +4777,7 @@ definitions:
           - "([-+]?\\d+)"
       max_waiting_time_in_seconds:
         title: Max Waiting Time in Seconds
-        description: Stop the stream instead of waiting, when the wait this strategy computes is longer than this value. The comparison is against the computed wait rather than the raw header, since the header holds an absolute timestamp, and it is applied after `min_wait`, so a cap below the floor still wins. Can be a hardcoded number or a string interpolated from the connector config, which lets an operation that must answer quickly, such as a connection check, use a tighter bound than a sync. A value of 0 means never wait.
+        description: Stop the stream instead of waiting, when the wait this strategy computes is longer than this value. The comparison is against the computed wait rather than the raw header, since the header holds an absolute timestamp, and it is applied after `min_wait`, so a cap below the floor still wins -- including for the fallback where the header is absent and `min_wait` supplies the wait on its own. Can be a hardcoded number, or a string interpolated from the connector config so that the bound can be changed without a connector release. A value of 0 means never wait.
         anyOf:
           - type: number
           - type: string
diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py
index 67888b241d..22c77cd6ab 100644
--- a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py
+++ b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py
@@ -1436,7 +1436,7 @@ class WaitTimeFromHeader(BaseModel):
     )
     max_waiting_time_in_seconds: Optional[Union[float, str]] = Field(
         None,
-        description="Stop the stream instead of waiting, when the value extracted from the header is greater than this value. Can be a hardcoded number or a string interpolated from the connector config, which lets an operation that must answer quickly, such as a connection check, use a tighter bound than a sync. A value of 0 means never wait.",
+        description="Stop the stream instead of waiting, when the value extracted from the header is greater than or equal to this value. Can be a hardcoded number, or a string interpolated from the connector config so that the bound can be changed without a connector release. A value of 0 means never wait.",
         examples=[3600, "{{ config['max_waiting_time'] * 60 }}"],
         title="Max Waiting Time in Seconds",
     )
@@ -1465,7 +1465,7 @@ class WaitUntilTimeFromHeader(BaseModel):
     )
     max_waiting_time_in_seconds: Optional[Union[float, str]] = Field(
         None,
-        description="Stop the stream instead of waiting, when the wait this strategy computes is longer than this value. The comparison is against the computed wait rather than the raw header, since the header holds an absolute timestamp, and it is applied after `min_wait`, so a cap below the floor still wins. Can be a hardcoded number or a string interpolated from the connector config, which lets an operation that must answer quickly, such as a connection check, use a tighter bound than a sync. A value of 0 means never wait.",
+        description="Stop the stream instead of waiting, when the wait this strategy computes is longer than this value. The comparison is against the computed wait rather than the raw header, since the header holds an absolute timestamp, and it is applied after `min_wait`, so a cap below the floor still wins -- including for the fallback where the header is absent and `min_wait` supplies the wait on its own. Can be a hardcoded number, or a string interpolated from the connector config so that the bound can be changed without a connector release. A value of 0 means never wait.",
         examples=[3600, "{{ config['max_waiting_time'] * 60 }}"],
         title="Max Waiting Time in Seconds",
     )
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 2512c34470..c9baa94e62 100644
--- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py
+++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py
@@ -4334,9 +4334,7 @@ def create_wait_time_from_header(
             parameters=model.parameters or {},
             config=config,
             regex=model.regex,
-            max_waiting_time_in_seconds=model.max_waiting_time_in_seconds
-            if model.max_waiting_time_in_seconds is not None
-            else None,
+            max_waiting_time_in_seconds=model.max_waiting_time_in_seconds,
         )
 
     @staticmethod
diff --git a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/max_waiting_time_helper.py b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/max_waiting_time_helper.py
new file mode 100644
index 0000000000..f59bda10ea
--- /dev/null
+++ b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/max_waiting_time_helper.py
@@ -0,0 +1,72 @@
+#
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+#
+
+from typing import Any, Mapping, Optional, Union
+
+from airbyte_cdk.models import FailureType
+from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString
+from airbyte_cdk.sources.types import Config
+from airbyte_cdk.utils import AirbyteTracedException
+
+MAX_WAITING_TIME_FIELD = "max_waiting_time_in_seconds"
+
+
+def interpolated_max_waiting_time(
+    max_waiting_time_in_seconds: Optional[Union[float, InterpolatedString, str]],
+    parameters: Mapping[str, Any],
+) -> Optional[InterpolatedString]:
+    """
+    Cast a `max_waiting_time_in_seconds` field to an InterpolatedString so that a hardcoded number
+    and a value interpolated from the config are resolved through the same path.
+
+    :param max_waiting_time_in_seconds: the value as declared on the backoff strategy
+    :param parameters: parameters to make available to the interpolation
+    :return: the value as an InterpolatedString, or None when no cap is configured
+    """
+    if max_waiting_time_in_seconds is None or isinstance(
+        max_waiting_time_in_seconds, InterpolatedString
+    ):
+        return max_waiting_time_in_seconds
+    return InterpolatedString.create(str(max_waiting_time_in_seconds), parameters=parameters)
+
+
+def evaluate_max_waiting_time(
+    max_waiting_time_in_seconds: Optional[InterpolatedString], config: Config
+) -> Optional[float]:
+    """
+    Resolve a `max_waiting_time_in_seconds` field to a number of seconds.
+
+    The `is None` checks are deliberate: 0 is a meaningful cap -- "never wait" -- so a truthiness
+    check would silently disable it.
+
+    A cap is only read while handling an error the requester is already going to retry, so an
+    interpolation that cannot be resolved would otherwise surface as an unhandled jinja
+    UndefinedError or ValueError in the middle of a sync that has been running fine. Raising a
+    config error instead names the field and points at the configuration that has to change.
+
+    :param max_waiting_time_in_seconds: the interpolated field, or None when no cap is configured
+    :param config: the connector config to interpolate against
+    :return: the cap in seconds, or None when no cap is configured
+    """
+    if max_waiting_time_in_seconds is None:
+        return None
+    try:
+        evaluated = max_waiting_time_in_seconds.eval(config)
+        if evaluated is None or evaluated == "":
+            return None
+        return float(evaluated)
+    except AirbyteTracedException:
+        raise
+    except Exception as exception:
+        raise AirbyteTracedException(
+            internal_message=(
+                f"Failed to evaluate {MAX_WAITING_TIME_FIELD} "
+                f"{max_waiting_time_in_seconds.string!r}: {exception}"
+            ),
+            message=(
+                f"The maximum rate limit waiting time is misconfigured. Check the value of "
+                f"`{MAX_WAITING_TIME_FIELD}` and the connector configuration it reads."
+            ),
+            failure_type=FailureType.config_error,
+        ) from exception
diff --git a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_time_from_header_backoff_strategy.py b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_time_from_header_backoff_strategy.py
index b8eb0cf5c0..294ffa7de8 100644
--- a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_time_from_header_backoff_strategy.py
+++ b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_time_from_header_backoff_strategy.py
@@ -13,6 +13,10 @@
 from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies.header_helper import (
     get_numeric_value_from_header,
 )
+from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies.max_waiting_time_helper import (
+    evaluate_max_waiting_time,
+    interpolated_max_waiting_time,
+)
 from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategy import (
     BackoffStrategy,
 )
@@ -28,7 +32,8 @@ class WaitTimeFromHeaderBackoffStrategy(BackoffStrategy):
     Attributes:
         header (str): header to read wait time from
         regex (Optional[str]): optional regex to apply on the header to extract its value
-        max_waiting_time_in_seconds: (Optional[float]): given the value extracted from the header is greater than this value, stop the stream
+        max_waiting_time_in_seconds (Optional[Union[float, InterpolatedString, str]]): stop the stream
+            rather than wait longer than this
     """
 
     header: Union[InterpolatedString, str]
@@ -42,13 +47,8 @@ def __post_init__(self, parameters: Mapping[str, Any]) -> None:
             InterpolatedString.create(self.regex, parameters=parameters) if self.regex else None
         )
         self.header = InterpolatedString.create(self.header, parameters=parameters)
-        self._max_waiting_time_in_seconds = (
-            self.max_waiting_time_in_seconds
-            if self.max_waiting_time_in_seconds is None
-            or isinstance(self.max_waiting_time_in_seconds, InterpolatedString)
-            else InterpolatedString.create(
-                str(self.max_waiting_time_in_seconds), parameters=parameters
-            )
+        self._max_waiting_time_in_seconds = interpolated_max_waiting_time(
+            self.max_waiting_time_in_seconds, parameters
         )
 
     def backoff_time(
@@ -65,25 +65,25 @@ def backoff_time(
         header_value = None
         if isinstance(response_or_exception, requests.Response):
             header_value = get_numeric_value_from_header(response_or_exception, header, regex)
-            max_waiting_time = self._eval_max_waiting_time()
-            # `is not None` rather than a truthiness check, so that 0 means "never wait" instead
-            # of silently disabling the cap.
+            max_waiting_time = evaluate_max_waiting_time(
+                self._max_waiting_time_in_seconds, self.config
+            )
+            # `max_waiting_time is not None` rather than a truthiness check, so that 0 means
+            # "never wait" instead of silently disabling the cap. The comparison stays `>=`,
+            # which is what this cap has always done; `WaitUntilTimeFromHeader` stops at `>`,
+            # so a wait exactly equal to the cap is allowed there and refused here.
             if (
                 max_waiting_time is not None
                 and header_value is not None
                 and header_value >= max_waiting_time
             ):
                 raise AirbyteTracedException(
-                    internal_message=f"Rate limit wait time {header_value} is greater than max waiting time of {max_waiting_time} seconds. Stopping the stream...",
-                    message="The rate limit is greater than max waiting time has been reached.",
+                    internal_message=(
+                        f"Rate limit wait time {header_value}s is greater than or equal to the "
+                        f"maximum of {max_waiting_time}s this stream is allowed to wait. "
+                        f"Stopping the stream..."
+                    ),
+                    message="The rate limit wait time is longer than the connector is allowed to wait.",
                     failure_type=FailureType.transient_error,
                 )
         return header_value
-
-    def _eval_max_waiting_time(self) -> Optional[float]:
-        if self._max_waiting_time_in_seconds is None:
-            return None
-        evaluated = self._max_waiting_time_in_seconds.eval(self.config)
-        if evaluated is None or evaluated == "":
-            return None
-        return float(evaluated)
diff --git a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_until_time_from_header_backoff_strategy.py b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_until_time_from_header_backoff_strategy.py
index 478269dd2a..663c03e7ff 100644
--- a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_until_time_from_header_backoff_strategy.py
+++ b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_until_time_from_header_backoff_strategy.py
@@ -2,7 +2,6 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
 
-import numbers
 import re
 import time
 from dataclasses import InitVar, dataclass
@@ -15,6 +14,10 @@
 from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies.header_helper import (
     get_numeric_value_from_header,
 )
+from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies.max_waiting_time_helper import (
+    evaluate_max_waiting_time,
+    interpolated_max_waiting_time,
+)
 from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategy import (
     BackoffStrategy,
 )
@@ -30,9 +33,10 @@ class WaitUntilTimeFromHeaderBackoffStrategy(BackoffStrategy):
 
     Attributes:
         header (str): header to read wait time from
-        min_wait (Optional[float]): minimum time to wait for safety
+        min_wait (Optional[Union[float, InterpolatedString, str]]): minimum time to wait for safety
         regex (Optional[str]): optional regex to apply on the header to extract its value
-        max_waiting_time_in_seconds (Optional[float]): stop the stream rather than wait longer than this
+        max_waiting_time_in_seconds (Optional[Union[float, InterpolatedString, str]]): stop the stream
+            rather than wait longer than this
     """
 
     header: Union[InterpolatedString, str]
@@ -49,13 +53,8 @@ def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         )
         if not isinstance(self.min_wait, InterpolatedString):
             self.min_wait = InterpolatedString.create(str(self.min_wait), parameters=parameters)
-        self._max_waiting_time_in_seconds = (
-            self.max_waiting_time_in_seconds
-            if self.max_waiting_time_in_seconds is None
-            or isinstance(self.max_waiting_time_in_seconds, InterpolatedString)
-            else InterpolatedString.create(
-                str(self.max_waiting_time_in_seconds), parameters=parameters
-            )
+        self._max_waiting_time_in_seconds = interpolated_max_waiting_time(
+            self.max_waiting_time_in_seconds, parameters
         )
 
     def backoff_time(
@@ -72,16 +71,12 @@ def backoff_time(
             regex = None
         wait_until = None
         if isinstance(response_or_exception, requests.Response):
+            # get_numeric_value_from_header returns a float or None, never a string
             wait_until = get_numeric_value_from_header(response_or_exception, header, regex)
         min_wait = self.min_wait.eval(self.config)  # type: ignore # header is always cast to an interpolated string
-        if wait_until is None or not wait_until:
+        if not wait_until:
             return self._capped(float(min_wait)) if min_wait else None
-        if (isinstance(wait_until, str) and wait_until.isnumeric()) or isinstance(
-            wait_until, numbers.Number
-        ):
-            wait_time = float(wait_until) - now
-        else:
-            return self._capped(float(min_wait))
+        wait_time = wait_until - now
         if min_wait:
             return self._capped(float(max(wait_time, min_wait)))
         elif wait_time < 0:
@@ -97,7 +92,7 @@ def _capped(self, wait_time: float) -> float:
         cap below the floor wins -- a caller asking never to wait more than N seconds means it,
         even when the floor would otherwise round the wait up past N.
         """
-        max_waiting_time = self._eval_max_waiting_time()
+        max_waiting_time = evaluate_max_waiting_time(self._max_waiting_time_in_seconds, self.config)
         if max_waiting_time is not None and wait_time > max_waiting_time:
             raise AirbyteTracedException(
                 internal_message=(
@@ -108,13 +103,3 @@ def _capped(self, wait_time: float) -> float:
                 failure_type=FailureType.transient_error,
             )
         return wait_time
-
-    def _eval_max_waiting_time(self) -> Optional[float]:
-        if self._max_waiting_time_in_seconds is None:
-            return None
-        evaluated = self._max_waiting_time_in_seconds.eval(self.config)
-        # `is None` rather than a truthiness check: 0 is a meaningful cap -- "never wait" -- and
-        # the equivalent field on WaitTimeFromHeader silently disables itself when set to 0.
-        if evaluated is None or evaluated == "":
-            return None
-        return float(evaluated)
diff --git a/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_header.py b/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_header.py
index d37766d120..c8a296c58b 100644
--- a/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_header.py
+++ b/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_header.py
@@ -77,3 +77,70 @@ def test_given_retry_after_greater_than_max_time_then_raise_transient_error():
     with pytest.raises(AirbyteTracedException) as exception:
         backoff_strategy.backoff_time(response_mock, 1)
     assert exception.value.failure_type == FailureType.transient_error
+
+
+def _response(header_value):
+    response = MagicMock(spec=Response)
+    response.headers = {_A_RETRY_HEADER: str(header_value)}
+    return response
+
+
+def _strategy(max_waiting_time_in_seconds, config=None):
+    return WaitTimeFromHeaderBackoffStrategy(
+        header=_A_RETRY_HEADER,
+        max_waiting_time_in_seconds=max_waiting_time_in_seconds,
+        parameters={},
+        config=config if config is not None else {},
+    )
+
+
+def test_given_max_waiting_time_is_zero_then_never_wait():
+    """`0` is the value a caller uses to say "never wait". It used to be read as falsy, which
+    silently disabled the cap; it is now honoured, which is the one behaviour change of the PR
+    that introduced interpolation on this field."""
+    strategy = _strategy(max_waiting_time_in_seconds=0)
+
+    with pytest.raises(AirbyteTracedException) as exc_info:
+        strategy.backoff_time(_response(1), 1)
+    assert exc_info.value.failure_type == FailureType.transient_error
+
+
+@pytest.mark.parametrize(
+    "config, expected",
+    [
+        pytest.param({"max_waiting_time": 10}, 120, id="cap_above_the_header_value_waits"),
+        pytest.param({"max_waiting_time": 1}, "raises", id="cap_below_the_header_value_raises"),
+        pytest.param({"max_waiting_time": 0}, "raises", id="zero_cap_never_waits"),
+    ],
+)
+def test_max_waiting_time_is_interpolated_from_config(config, expected):
+    strategy = _strategy("{{ config['max_waiting_time'] * 60 }}", config=config)
+
+    if expected == "raises":
+        with pytest.raises(AirbyteTracedException) as exc_info:
+            strategy.backoff_time(_response(120), 1)
+        assert exc_info.value.failure_type == FailureType.transient_error
+    else:
+        assert strategy.backoff_time(_response(120), 1) == expected
+
+
+@pytest.mark.parametrize(
+    "max_waiting_time_in_seconds, config",
+    [
+        pytest.param("{{ config['max_waiting_time'] * 60 }}", {}, id="config_value_is_missing"),
+        pytest.param(
+            "{{ config['max_waiting_time'] }}", {"max_waiting_time": "abc"}, id="not_a_number"
+        ),
+    ],
+)
+def test_given_max_waiting_time_cannot_be_evaluated_then_raise_config_error(
+    max_waiting_time_in_seconds, config
+):
+    """The cap is only read while handling an error that was already going to be retried, so an
+    unresolvable interpolation must not surface as an unhandled jinja or float error."""
+    strategy = _strategy(max_waiting_time_in_seconds, config=config)
+
+    with pytest.raises(AirbyteTracedException) as exc_info:
+        strategy.backoff_time(_response(120), 1)
+    assert exc_info.value.failure_type == FailureType.config_error
+    assert "max_waiting_time_in_seconds" in exc_info.value.internal_message
diff --git a/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_until_time_from_header.py b/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_until_time_from_header.py
index e6436ac88e..ef0ed9f124 100644
--- a/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_until_time_from_header.py
+++ b/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_until_time_from_header.py
@@ -214,3 +214,26 @@ def test_cap_is_interpolated_from_config(time_mock, config, expected):
             strategy.backoff_time(_response(), 1)
     else:
         assert strategy.backoff_time(_response(), 1) == expected
+
+
+@pytest.mark.parametrize(
+    "max_waiting_time_in_seconds, config",
+    [
+        pytest.param("{{ config['max_waiting_time'] * 60 }}", {}, id="config_value_is_missing"),
+        pytest.param(
+            "{{ config['max_waiting_time'] }}", {"max_waiting_time": "abc"}, id="not_a_number"
+        ),
+    ],
+)
+@patch("time.time", return_value=NOW)
+def test_given_cap_cannot_be_evaluated_then_raise_config_error(
+    time_mock, max_waiting_time_in_seconds, config
+):
+    """The cap is only read while handling an error that was already going to be retried, so an
+    unresolvable interpolation must not surface as an unhandled jinja or float error."""
+    strategy = _strategy(max_waiting_time_in_seconds, config=config)
+
+    with pytest.raises(AirbyteTracedException) as exc_info:
+        strategy.backoff_time(_response(), 1)
+    assert exc_info.value.failure_type == FailureType.config_error
+    assert "max_waiting_time_in_seconds" in exc_info.value.internal_message

From 94a0cc2b282fff8d196af7cdf18e3cbe0aab3b76 Mon Sep 17 00:00:00 2001
From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com>
Date: Wed, 19 Aug 2026 17:48:06 +0300
Subject: [PATCH 3/5] fix(low-code): raise a system error when the wait cap
 cannot be evaluated

`max_waiting_time_in_seconds` is declared in the manifest, so a cap that
cannot be resolved is the connector's fault, not the user's: either the
expression is wrong, or it reads a config key the spec does not expose. Either
way there is nothing in the connector settings for the user to correct, so a
config error pointed them at a field they cannot see.

The message now says the connector could not determine its wait budget and
that the configuration is not at fault; the field name and the offending
expression stay in the internal message. An AirbyteTracedException raised by
the interpolation itself still passes through untouched, keeping its own
failure type and message.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../backoff_strategies/max_waiting_time_helper.py   | 13 ++++++++-----
 .../test_wait_time_from_header.py                   |  7 ++++---
 .../test_wait_until_time_from_header.py             |  7 ++++---
 3 files changed, 16 insertions(+), 11 deletions(-)

diff --git a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/max_waiting_time_helper.py b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/max_waiting_time_helper.py
index f59bda10ea..710c718536 100644
--- a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/max_waiting_time_helper.py
+++ b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/max_waiting_time_helper.py
@@ -42,8 +42,10 @@ def evaluate_max_waiting_time(
 
     A cap is only read while handling an error the requester is already going to retry, so an
     interpolation that cannot be resolved would otherwise surface as an unhandled jinja
-    UndefinedError or ValueError in the middle of a sync that has been running fine. Raising a
-    config error instead names the field and points at the configuration that has to change.
+    UndefinedError or ValueError in the middle of a sync that has been running fine. It is raised
+    as a system error rather than a config error because the field lives in the manifest: whether
+    it is a bad expression or a config key the manifest reads but the spec does not expose, the
+    connector is at fault and there is nothing for the user to correct.
 
     :param max_waiting_time_in_seconds: the interpolated field, or None when no cap is configured
     :param config: the connector config to interpolate against
@@ -65,8 +67,9 @@ def evaluate_max_waiting_time(
                 f"{max_waiting_time_in_seconds.string!r}: {exception}"
             ),
             message=(
-                f"The maximum rate limit waiting time is misconfigured. Check the value of "
-                f"`{MAX_WAITING_TIME_FIELD}` and the connector configuration it reads."
+                "The connector could not determine how long it is allowed to wait for a rate "
+                "limit to clear. This is a problem with the connector rather than with your "
+                "configuration."
             ),
-            failure_type=FailureType.config_error,
+            failure_type=FailureType.system_error,
         ) from exception
diff --git a/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_header.py b/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_header.py
index c8a296c58b..4879be27c0 100644
--- a/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_header.py
+++ b/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_header.py
@@ -133,14 +133,15 @@ def test_max_waiting_time_is_interpolated_from_config(config, expected):
         ),
     ],
 )
-def test_given_max_waiting_time_cannot_be_evaluated_then_raise_config_error(
+def test_given_max_waiting_time_cannot_be_evaluated_then_raise_system_error(
     max_waiting_time_in_seconds, config
 ):
     """The cap is only read while handling an error that was already going to be retried, so an
-    unresolvable interpolation must not surface as an unhandled jinja or float error."""
+    unresolvable interpolation must not surface as an unhandled jinja or float error. It is a
+    system error because the field is declared in the manifest: the user has nothing to fix."""
     strategy = _strategy(max_waiting_time_in_seconds, config=config)
 
     with pytest.raises(AirbyteTracedException) as exc_info:
         strategy.backoff_time(_response(120), 1)
-    assert exc_info.value.failure_type == FailureType.config_error
+    assert exc_info.value.failure_type == FailureType.system_error
     assert "max_waiting_time_in_seconds" in exc_info.value.internal_message
diff --git a/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_until_time_from_header.py b/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_until_time_from_header.py
index ef0ed9f124..79a013cb40 100644
--- a/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_until_time_from_header.py
+++ b/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_until_time_from_header.py
@@ -226,14 +226,15 @@ def test_cap_is_interpolated_from_config(time_mock, config, expected):
     ],
 )
 @patch("time.time", return_value=NOW)
-def test_given_cap_cannot_be_evaluated_then_raise_config_error(
+def test_given_cap_cannot_be_evaluated_then_raise_system_error(
     time_mock, max_waiting_time_in_seconds, config
 ):
     """The cap is only read while handling an error that was already going to be retried, so an
-    unresolvable interpolation must not surface as an unhandled jinja or float error."""
+    unresolvable interpolation must not surface as an unhandled jinja or float error. It is a
+    system error because the field is declared in the manifest: the user has nothing to fix."""
     strategy = _strategy(max_waiting_time_in_seconds, config=config)
 
     with pytest.raises(AirbyteTracedException) as exc_info:
         strategy.backoff_time(_response(), 1)
-    assert exc_info.value.failure_type == FailureType.config_error
+    assert exc_info.value.failure_type == FailureType.system_error
     assert "max_waiting_time_in_seconds" in exc_info.value.internal_message

From 1b8c9811ed2f80a343979587235924ae9a725185 Mon Sep 17 00:00:00 2001
From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com>
Date: Wed, 19 Aug 2026 17:55:42 +0300
Subject: [PATCH 4/5] fix(low-code): let a zero header through the cap, and
 stop dropping a cap that resolves to nothing

Two review follow-ups on `max_waiting_time_in_seconds`:

- `WaitTimeFromHeader` checked `header_value is not None`, so with a cap of 0
  a `Retry-After: 0` response stopped the stream over a wait of zero seconds.
  Headers arrive as strings and `"0"` reads as 0.0, not as a missing header,
  so the branch was reachable in production even though only an integer 0 --
  a mock -- yields None. Back to a truthiness check: a header asking for no
  wait is not a wait any cap should refuse.
- A cap that resolved to nothing was treated as "no cap", so a blank config
  value silently restored the unbounded wait the field exists to prevent,
  while a whitespace value and a missing key both raised. "No cap" is already
  spelled by leaving the field out of the manifest, so a field that is present
  and resolves to nothing now raises like any other unusable value.

Tests: a zero header is allowed through with a cap of 0 and with a finite cap;
an empty and a null config value each raise; and the AirbyteTracedException
passthrough is pinned, so interpolation errors keep their own failure type and
message instead of being reclassified as system errors.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../max_waiting_time_helper.py                | 13 +++---
 .../wait_time_from_header_backoff_strategy.py |  8 ++--
 .../test_wait_time_from_header.py             | 40 +++++++++++++++++++
 3 files changed, 50 insertions(+), 11 deletions(-)

diff --git a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/max_waiting_time_helper.py b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/max_waiting_time_helper.py
index 710c718536..280dab103d 100644
--- a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/max_waiting_time_helper.py
+++ b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/max_waiting_time_helper.py
@@ -37,8 +37,9 @@ def evaluate_max_waiting_time(
     """
     Resolve a `max_waiting_time_in_seconds` field to a number of seconds.
 
-    The `is None` checks are deliberate: 0 is a meaningful cap -- "never wait" -- so a truthiness
-    check would silently disable it.
+    The `is None` check is deliberate: 0 is a meaningful cap -- "never wait" -- so a truthiness
+    check would silently disable it. Only an absent field means "no cap"; a field that is present
+    but resolves to nothing raises, rather than quietly leaving the wait unbounded.
 
     A cap is only read while handling an error the requester is already going to retry, so an
     interpolation that cannot be resolved would otherwise surface as an unhandled jinja
@@ -54,10 +55,10 @@ def evaluate_max_waiting_time(
     if max_waiting_time_in_seconds is None:
         return None
     try:
-        evaluated = max_waiting_time_in_seconds.eval(config)
-        if evaluated is None or evaluated == "":
-            return None
-        return float(evaluated)
+        # A cap that resolves to nothing -- an empty or null config value -- is a failure rather
+        # than "no cap": silently dropping the bound restores the unbounded wait the field exists
+        # to prevent, and "no cap" is already spelled by leaving the field out of the manifest.
+        return float(max_waiting_time_in_seconds.eval(config))
     except AirbyteTracedException:
         raise
     except Exception as exception:
diff --git a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_time_from_header_backoff_strategy.py b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_time_from_header_backoff_strategy.py
index 294ffa7de8..b62b013417 100644
--- a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_time_from_header_backoff_strategy.py
+++ b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_time_from_header_backoff_strategy.py
@@ -72,11 +72,9 @@ def backoff_time(
             # "never wait" instead of silently disabling the cap. The comparison stays `>=`,
             # which is what this cap has always done; `WaitUntilTimeFromHeader` stops at `>`,
             # so a wait exactly equal to the cap is allowed there and refused here.
-            if (
-                max_waiting_time is not None
-                and header_value is not None
-                and header_value >= max_waiting_time
-            ):
+            # `header_value` is checked for truthiness rather than `is not None` on purpose: a
+            # header of `0` asks for no wait at all, which no cap -- not even 0 -- should refuse.
+            if max_waiting_time is not None and header_value and header_value >= max_waiting_time:
                 raise AirbyteTracedException(
                     internal_message=(
                         f"Rate limit wait time {header_value}s is greater than or equal to the "
diff --git a/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_header.py b/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_header.py
index 4879be27c0..29912086c5 100644
--- a/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_header.py
+++ b/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_header.py
@@ -145,3 +145,43 @@ def test_given_max_waiting_time_cannot_be_evaluated_then_raise_system_error(
         strategy.backoff_time(_response(120), 1)
     assert exc_info.value.failure_type == FailureType.system_error
     assert "max_waiting_time_in_seconds" in exc_info.value.internal_message
+
+
+@pytest.mark.parametrize(
+    "max_waiting_time_in_seconds",
+    [pytest.param(0, id="never_wait"), pytest.param(60, id="finite_cap")],
+)
+def test_given_header_asks_for_no_wait_then_no_cap_refuses_it(max_waiting_time_in_seconds):
+    """`Retry-After: 0` asks for no wait at all, so no cap -- not even 0 -- should stop the stream
+    over it. Headers arrive as strings, and `"0"` reads as 0.0 rather than as a missing header."""
+    strategy = _strategy(max_waiting_time_in_seconds)
+
+    assert strategy.backoff_time(_response(0), 1) == 0
+
+
+@pytest.mark.parametrize(
+    "config",
+    [
+        pytest.param({"max_waiting_time": ""}, id="config_value_is_empty"),
+        pytest.param({"max_waiting_time": None}, id="config_value_is_null"),
+    ],
+)
+def test_given_max_waiting_time_resolves_to_nothing_then_raise_rather_than_drop_the_cap(config):
+    """A blank config value must not leave the wait unbounded: "no cap" is spelled by leaving the
+    field out of the manifest, so a field that is present and resolves to nothing is a failure."""
+    strategy = _strategy("{{ config['max_waiting_time'] }}", config=config)
+
+    with pytest.raises(AirbyteTracedException) as exc_info:
+        strategy.backoff_time(_response(120), 1)
+    assert exc_info.value.failure_type == FailureType.system_error
+
+
+def test_given_interpolation_raises_a_traced_error_then_keep_its_own_failure_type_and_message():
+    """`stream_state` interpolation raises an AirbyteTracedException of its own, with a message
+    written for that case. The cap's own error handling must not reclassify or replace it."""
+    strategy = _strategy("{{ stream_state['max_waiting_time'] }}")
+
+    with pytest.raises(AirbyteTracedException) as exc_info:
+        strategy.backoff_time(_response(120), 1)
+    assert exc_info.value.failure_type == FailureType.config_error
+    assert "`stream_state` is no longer supported for interpolation" in exc_info.value.message

From cd44bc3becdafad080d6921820d2d67e1805a7ed Mon Sep 17 00:00:00 2001
From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com>
Date: Wed, 19 Aug 2026 18:39:41 +0300
Subject: [PATCH 5/5] fix(low-code): reject a non-finite cap, and align the
 boundary with the sibling

Two review findings, both real.

NaN is the one value that would switch the cap off without saying so: every
comparison against it is False, so the wait this field exists to bound would run
unbounded again -- the exact outcome the helper's docstring already argues must
not happen silently. It is now rejected, along with infinity as the same kind of
mistake; "no cap" stays spelled by leaving the field out.

The cap boundary was `>` here and `>=` on WaitTimeFromHeader, so one field name
meant two different things depending on which strategy it was written on. Aligned
on the released `>=`, which also makes a cap of 0 refuse every wait rather than
only waits above zero -- what "never wait" has to mean.

Both guards were mutation-checked: reverting the boundary and removing the
isfinite check each fail their tests.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../max_waiting_time_helper.py                |  9 +++++++-
 ...until_time_from_header_backoff_strategy.py |  5 +++-
 .../test_wait_time_from_header.py             |  1 +
 .../test_wait_until_time_from_header.py       | 23 ++++++++++++++++++-
 4 files changed, 35 insertions(+), 3 deletions(-)

diff --git a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/max_waiting_time_helper.py b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/max_waiting_time_helper.py
index 280dab103d..82feb3c976 100644
--- a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/max_waiting_time_helper.py
+++ b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/max_waiting_time_helper.py
@@ -2,6 +2,7 @@
 # Copyright (c) 2025 Airbyte, Inc., all rights reserved.
 #
 
+import math
 from typing import Any, Mapping, Optional, Union
 
 from airbyte_cdk.models import FailureType
@@ -58,7 +59,13 @@ def evaluate_max_waiting_time(
         # A cap that resolves to nothing -- an empty or null config value -- is a failure rather
         # than "no cap": silently dropping the bound restores the unbounded wait the field exists
         # to prevent, and "no cap" is already spelled by leaving the field out of the manifest.
-        return float(max_waiting_time_in_seconds.eval(config))
+        max_waiting_time = float(max_waiting_time_in_seconds.eval(config))
+        if not math.isfinite(max_waiting_time):
+            # NaN would be the one value that disables the cap without saying so: every
+            # comparison against it is False, so the wait this field exists to bound would run
+            # unbounded again. Infinity is rejected alongside it as the same kind of mistake.
+            raise ValueError(f"resolved to {max_waiting_time}, which is not a finite number")
+        return max_waiting_time
     except AirbyteTracedException:
         raise
     except Exception as exception:
diff --git a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_until_time_from_header_backoff_strategy.py b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_until_time_from_header_backoff_strategy.py
index 663c03e7ff..8d419060a3 100644
--- a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_until_time_from_header_backoff_strategy.py
+++ b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_until_time_from_header_backoff_strategy.py
@@ -93,7 +93,10 @@ def _capped(self, wait_time: float) -> float:
         even when the floor would otherwise round the wait up past N.
         """
         max_waiting_time = evaluate_max_waiting_time(self._max_waiting_time_in_seconds, self.config)
-        if max_waiting_time is not None and wait_time > max_waiting_time:
+        # `>=` rather than `>` to match WaitTimeFromHeader, so one field name does not mean two
+        # different things depending on which strategy it is written on. A cap of 0 therefore
+        # refuses every wait, which is what "never wait" has to mean.
+        if max_waiting_time is not None and wait_time >= max_waiting_time:
             raise AirbyteTracedException(
                 internal_message=(
                     f"Rate limit wait time {wait_time}s is greater than the maximum of "
diff --git a/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_header.py b/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_header.py
index 29912086c5..35c1ece797 100644
--- a/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_header.py
+++ b/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_header.py
@@ -109,6 +109,7 @@ def test_given_max_waiting_time_is_zero_then_never_wait():
     "config, expected",
     [
         pytest.param({"max_waiting_time": 10}, 120, id="cap_above_the_header_value_waits"),
+        pytest.param({"max_waiting_time": 2}, "raises", id="cap_equal_to_the_header_value_raises"),
         pytest.param({"max_waiting_time": 1}, "raises", id="cap_below_the_header_value_raises"),
         pytest.param({"max_waiting_time": 0}, "raises", id="zero_cap_never_waits"),
     ],
diff --git a/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_until_time_from_header.py b/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_until_time_from_header.py
index 79a013cb40..fdaa0d890e 100644
--- a/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_until_time_from_header.py
+++ b/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_until_time_from_header.py
@@ -151,7 +151,8 @@ def _strategy(max_waiting_time_in_seconds, min_wait=None, config=None):
     [
         pytest.param(None, 60, id="no_cap_waits"),
         pytest.param(3600, 60, id="cap_above_the_wait_waits"),
-        pytest.param(60, 60, id="cap_equal_to_the_wait_waits"),
+        pytest.param(60, "raises", id="cap_equal_to_the_wait_raises"),
+        pytest.param(61, 60, id="cap_just_above_the_wait_waits"),
         pytest.param(30, "raises", id="cap_below_the_wait_raises"),
         pytest.param(0, "raises", id="zero_cap_never_waits"),
     ],
@@ -162,6 +163,9 @@ def test_max_waiting_time_in_seconds(time_mock, max_waiting_time_in_seconds, exp
 
     `0` has to raise rather than switch the cap off: it is the value a caller uses to say "never
     wait", and the equivalent field on WaitTimeFromHeader read it as falsy and ignored it.
+
+    The boundary is `>=`, matching WaitTimeFromHeader, so that one field name does not mean two
+    different things depending on which strategy it is written on.
     """
     strategy = _strategy(max_waiting_time_in_seconds)
 
@@ -238,3 +242,20 @@ def test_given_cap_cannot_be_evaluated_then_raise_system_error(
         strategy.backoff_time(_response(), 1)
     assert exc_info.value.failure_type == FailureType.system_error
     assert "max_waiting_time_in_seconds" in exc_info.value.internal_message
+
+
+@pytest.mark.parametrize(
+    "cap",
+    [pytest.param("nan", id="nan"), pytest.param("inf", id="infinity")],
+)
+@patch("time.time", return_value=NOW)
+def test_non_finite_cap_is_rejected(time_mock, cap):
+    """NaN is the one value that would switch the cap off without saying so -- every comparison
+    against it is False, so the wait this field exists to bound would run unbounded again.
+    Infinity is rejected as the same kind of mistake rather than read as "no cap", which is
+    already spelled by leaving the field out."""
+    strategy = _strategy(cap)
+
+    with pytest.raises(AirbyteTracedException) as exc_info:
+        strategy.backoff_time(_response(), 1)
+    assert exc_info.value.failure_type == FailureType.system_error