feat(declarative): add WaitTimeFromResponse backoff strategy

[devin-ai-integration]

Summary

Adds a WaitTimeFromResponse backoff strategy so a manifest can compute the wait from the response body without a custom component. This covers APIs that only send the wait time in the body, such as Gmail: {"error": {"message": "... Retry after 2026-10-15T13:15:00Z"}}, with no Retry-After header.

WaitTimeFromResponseBackoffStrategy:

• wait_time_in_seconds: an interpolated expression with config, response (the parsed JSON body, {} when the body isn't JSON or the input is an exception) and headers.
• An empty, negative, non-finite or non-numeric result returns None, so DefaultErrorHandler moves on to the next strategy. So does an expression that cannot be evaluated on the response body (a list, or a string where an object was expected): the strategy runs on every retryable response, not only the one the expression was written for. A warning names the strategy and the error type, never the body.
• Optional max_waiting_time_in_seconds, with the same helper and semantics as WaitTimeFromHeader: if the wait is >= the cap, it raises transient_error. The cap is validated at construction.
• Registered in the schema, the DefaultErrorHandler.backoff_strategies union and ModelToComponentFactory.

Example (Gmail):

backoff_strategies:
  - type: WaitTimeFromHeader
    header: Retry-After
  - type: WaitTimeFromResponse
    wait_time_in_seconds: "{{ max(1, timestamp(response.error.message | regex_search('Retry after (\\S+Z)')) - now_utc().timestamp()) if (response is mapping and response.get('error') is mapping and ((response.error.message or '') | string | regex_search('Retry after (\\S+Z)'))) else '' }}"
    max_waiting_time_in_seconds: 1800
  - type: ExponentialBackoffStrategy   # used when the body has no timestamp
    factor: 5

The generated model in declarative_component_schema.py was edited by hand, because running poe assemble locally pulled in unrelated generator drift.

Tests: unit_tests/sources/declarative/requesters/error_handlers plus a factory test (518 passed together with test_model_to_component_factory.py). They cover the cap, empty and missing results, a non-JSON body, negative, NaN, infinite and non-numeric results, the floor-less retry_at - now expression with a past and a future time, and the Gmail example on five other body shapes.

Link to Devin session: https://app.devin.ai/sessions/4b1e021516004e0ab53c25b7293450a2
Open in Devin Desktop: https://app.devin.ai/desktop/session/4b1e021516004e0ab53c25b7293450a2?variant=devin

[devin-ai-integration]

I'll fix CI failures and address comments from users with write access. I'll skip comments containing "(aside)".

• Disable automatic comment, CI, and merge conflict monitoring

[github-actions]

👋 Greetings, Airbyte Team Member!

Here are some helpful tips and reminders for your convenience.

💡 Show Tips and Tricks

Testing This CDK Version

You can test this version of the CDK using the following:

# Run the CLI from this branch:
uvx 'git+https://github.com/airbytehq/airbyte-python-cdk.git@devin/1792070042-constant-backoff-response-interpolation#egg=airbyte-python-cdk[dev]' --help

# Update a connector to use the CDK from this branch ref:
cd airbyte-integrations/connectors/source-example
poe use-cdk-branch devin/1792070042-constant-backoff-response-interpolation

PR Slash Commands

Airbyte Maintainers can execute the following slash commands on your PR:

• /autofix - Fixes most formatting and linting issues
• /poetry-lock - Updates poetry.lock file
• /test - Runs connector tests with the updated CDK
• /prerelease - Triggers a prerelease publish with default arguments
• /poe build - Regenerate git-committed build artifacts, such as the pydantic models which are generated from the manifest JSON schema in YAML.
• /poe <command> - Runs any poe command in the CDK environment

📚 Show Repo Guidance

Helpful Resources

• CDK API Reference

📝 Edit this welcome message.

[github-actions]

PyTest Results (Fast)

5 785 tests  +31   5 754 ✅ +31   12m 5s ⏱️ +26s
    1 suites ± 0      31 💤 ± 0 
    1 files   ± 0       0 ❌ ± 0 

Results for commit d30f834. ± Comparison against base commit 73ccaa9.

♻️ This comment has been updated with latest results.

[github-actions]

PyTest Results (Full)

5 788 tests  +31   5 757 ✅ +31   17m 7s ⏱️ + 6m 59s
    1 suites ± 0      31 💤 ± 0 
    1 files   ± 0       0 ❌ ± 0 

Results for commit d30f834. ± Comparison against base commit 73ccaa9.

♻️ This comment has been updated with latest results.

[Anatolii Yatsuk (tolik0)]

Review of the new WaitTimeFromResponse strategy. The change is additive and opt-in, and I confirmed the hand-edited model matches datamodel-codegen 0.26.3 output exactly. One P1 (an odd response body can turn the strategy off for the rest of the sync) and four P2s are inline below.

[Anatolii Yatsuk (tolik0)]

P1: one odd response body turns this strategy off for the rest of the sync

InterpolatedString caches whether a template is plain text on its first eval: self._is_plain_string = self.string == evaluated (interpolation/interpolated_string.py:51). When rendering raises TypeError, JinjaInterpolation._eval catches it and returns the raw template unchanged (interpolation/jinja.py:145). The first evaluation then equals self.string, the template is cached as plain text, and every later call returns the literal "{{ ... }}". float() fails on it, and the strategy returns None from then on.

Reproduced at this head:

{{ response.reset_in - 1 }}
  body {"reset_in": "abc"} -> None   (_is_plain_string is now True)
  body {"reset_in": 30}    -> None   (expected 29)

{{ timestamp(response.retry_at) - now_utc().timestamp() }}   # the floor-less expression from the tests
  body {"retry_at": null}                    -> None   (_is_plain_string is now True)
  body {"retry_at": "2099-01-01T00:00:00Z"}  -> None

On the concurrent path one strategy instance is shared by every partition of a stream. A single retryable response with an unexpected shape, such as a 5xx with a different error body, therefore disables the strategy for the whole sync, and the only trace is a WARN line. For Gmail, the requests would then fall back to exponential backoff, which is much shorter than the server's "retry after" time.

The cache is existing InterpolatedString behaviour, but this is the first component that feeds arbitrary response bodies through it. Suggestion: evaluate with JinjaInterpolation directly, as InterpolatedBoolean does, and treat a result equal to the raw template as an evaluation failure:

# __post_init__
self._interpolation = JinjaInterpolation()
self._parameters = parameters

# backoff_time
evaluated = self._interpolation.eval(
    self._wait_time_template, self.config,
    parameters=self._parameters, response=response, headers=headers,
)
if evaluated == self._wait_time_template:  # render raised TypeError inside JinjaInterpolation._eval
    LOGGER.warning("... could not be evaluated (TypeError); deferring to the next backoff strategy.")
    return None

Please also add a regression test that reuses one instance: {"reset_in": "abc"} returns None, then {"reset_in": 30} returns 29. That test fails on the current head.

[Anatolii Yatsuk (tolik0)]

P2: what this except catches does not match the errors that can actually happen

• ValueError also catches the undeclared-variable check in JinjaInterpolation._eval (jinja.py:140-142). A typo such as {{ resposne.x }} makes the strategy do nothing for every response, with only a WARN line to show for it. That is a manifest bug and should fail at startup. Checking the template's undeclared variables in __post_init__ against config, response, headers and parameters would catch it there.
• The TypeError arm can never be reached: JinjaInterpolation._eval swallows render-time TypeError and returns the raw template (see the comment on line 44).
• ZeroDivisionError escapes and fails the stream, for example {{ 10 / response.n }} with {"n": 0} (reproduced). float() at line 88 can also raise OverflowError on a very large integer, and that is not in (TypeError, ValueError) either. Catching ArithmeticError in both places covers both.
• At lines 61-62 only JSONDecodeError is caught around .json(). A deeply nested body raises RecursionError, and with a stream=True decoder a broken chunked body raises ChunkedEncodingError. Both fail the stream, even when the expression reads only headers. Falling back to {} on (ValueError, requests.RequestException, RecursionError) would make the body read as tolerant as the evaluation.

[Anatolii Yatsuk (tolik0)]

P2: nothing bounds the wait when no max_waiting_time_in_seconds is set

Any finite positive value reaches time.sleep as it is. max_time does not shorten a sleep that has already started, because the sleep runs in the on_backoff handler (rate_limiting.py:110-119). Response bodies often carry absolute epoch or millisecond timestamps rather than relative seconds, so a one-character manifest mistake goes a long way:

• {{ response.reset }} with {"reset": 1760500000} sleeps for about 55 years, until the heartbeat kills the job.
• A value of about 9.2e9 or more makes time.sleep raise OverflowError: timestamp out of range for platform time_t, an untraced failure.

WaitTimeFromHeader has the same pattern, but headers rarely carry epochs. At minimum, please say in the field description that the result must be relative seconds and that a cap is strongly recommended. A hard ceiling here (return None with a warning above, say, a day) would be safer still.

[Anatolii Yatsuk (tolik0)]

P2: the descriptions leave out three things a manifest author needs to know

• Body format: only a JSON body is read (.json(), falling back to {}). With an XML, CSV or JSONL requester, response is always {}. Suggested wording: "Compute the wait time, in relative seconds, from the JSON response body or the response headers. A non-JSON body is exposed as an empty response."
• Expressions that can't be evaluated: this description lists only an empty result as deferring. The wait_time_in_seconds description covers empty, negative and non-numeric results, but not an expression that fails to evaluate on the body. The code defers in that case too.
• Zero waits: the code comment at wait_time_from_response_backoff_strategy.py:101 says "A zero wait is returned as 0". DefaultErrorHandler.backoff_time uses if backoff: (default_error_handler.py:163), so a 0 also moves on to the next strategy, unless this one is last.

The generated model description needs the same change.

[Anatolii Yatsuk (tolik0)]

P2: tests don't cover the strategy end to end, and this factory test checks only the type

This test asserts the type and the raw cap. Dropping parameters, or wiring wait_time_in_seconds wrongly, in create_wait_time_from_response would still pass. Calling backoff_strategy.backoff_time(<429 with {"retry_after_seconds": 12}>, 1) == 12 here, and adding a $parameters case, would close that gap.

Nothing sends a retryable response through HttpRequester/HttpClient either, so no test proves the computed wait actually reaches the sleep or works with a RATE_LIMITED filter. Suggested test: an HttpMocker manifest stream that returns a 429 with {"retry_in": 7} and then a 200, with time.sleep patched. Assert a sleep of about 7 and that records were read. Add a cap-exceeded case that asserts a transient_error trace.

For reference, a mutation run found that removing the TypeError arm, the ValueError arm, or the empty-string check in the strategy fails no test.