diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml
index 091c694d2..0a51d1e71 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 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
+        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 -- 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
+        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 37eb50b7b..22c77cd6a 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 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",
     )
     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 -- 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",
+    )
     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 fbecdacb3..f70d12c6f 100644
--- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py
+++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py
@@ -4343,9 +4343,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
@@ -4358,6 +4356,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/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 000000000..82feb3c97
--- /dev/null
+++ b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/max_waiting_time_helper.py
@@ -0,0 +1,83 @@
+#
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+#
+
+import math
+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` 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
+    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
+    :return: the cap in seconds, or None when no cap is configured
+    """
+    if max_waiting_time_in_seconds is None:
+        return None
+    try:
+        # 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.
+        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:
+        raise AirbyteTracedException(
+            internal_message=(
+                f"Failed to evaluate {MAX_WAITING_TIME_FIELD} "
+                f"{max_waiting_time_in_seconds.string!r}: {exception}"
+            ),
+            message=(
+                "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.system_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 5cda96a4d..b62b01341 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,20 +32,24 @@ 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]
     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 = interpolated_max_waiting_time(
+            self.max_waiting_time_in_seconds, parameters
+        )
 
     def backoff_time(
         self,
@@ -57,14 +65,23 @@ 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)
-            if (
-                self.max_waiting_time_in_seconds
-                and header_value
-                and header_value >= self.max_waiting_time_in_seconds
-            ):
+            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.
+            # `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} is greater than max waiting time of {self.max_waiting_time_in_seconds} 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
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 1220e198f..8d419060a 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
@@ -10,14 +9,20 @@
 
 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,
 )
+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,
 )
 from airbyte_cdk.sources.types import Config
+from airbyte_cdk.utils import AirbyteTracedException
 
 
 @dataclass
@@ -28,8 +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[Union[float, InterpolatedString, str]]): stop the stream
+            rather than wait longer than this
     """
 
     header: Union[InterpolatedString, str]
@@ -37,6 +44,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 +53,9 @@ 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 = interpolated_max_waiting_time(
+            self.max_waiting_time_in_seconds, parameters
+        )
 
     def backoff_time(
         self,
@@ -60,18 +71,38 @@ 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:
-            return 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)
+        if not wait_until:
+            return self._capped(float(min_wait)) if min_wait else None
+        wait_time = wait_until - now
         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 = evaluate_max_waiting_time(self._max_waiting_time_in_seconds, self.config)
+        # `>=` 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 "
+                    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
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 d37766d12..35c1ece79 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,112 @@ 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": 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"),
+    ],
+)
+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_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. 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.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
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 4c4c5a6f7..fdaa0d890 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,138 @@ 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, "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"),
+    ],
+)
+@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.
+
+    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)
+
+    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
+
+
+@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_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. 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.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