fix(low-code): preserve check config_overrides during parameter propagation

airbyte_cdk/sources/declarative/concurrent_declarative_source.py

@@ -3,6 +3,7 @@
 import json
 import logging
 import pkgutil
+from contextlib import contextmanager
 from copy import deepcopy
 from dataclasses import dataclass, field
 from queue import Queue
@@ -216,6 +217,8 @@ def __init__(
             AlwaysLogSliceLogger() if emit_connector_builder_messages else DebugSliceLogger()
         )
 
+        # Populated by `_pre_process_manifest` from the check component, before parameter propagation.
+        self._check_config_overrides: Optional[Mapping[str, Any]] = None
         # resolve all components in the manifest
         self._source_config = self._pre_process_manifest(dict(source_config))
         # validate resolved manifest against the declarative component schema
@@ -228,6 +231,10 @@ def __init__(
             self._constructor.create_component(SpecModel, spec, dict()) if spec else None
         )
         self._config = self._migrate_and_transform_config(config_path, config) or {}
+        # `check` may temporarily overlay values onto `self._config` (see
+        # `_config_overridden_for_check`). The manifest's `config_validations` express intent about the
+        # config as supplied, so they must run against it rather than against a check-time overlay.
+        self._config_for_validation = self._config
 
         concurrency_level_from_manifest = self._source_config.get("concurrency_level")
         if concurrency_level_from_manifest:
@@ -275,13 +282,38 @@ def _pre_process_manifest(self, manifest: Dict[str, Any]) -> Dict[str, Any]:
         manifest = self._fix_source_type(manifest)
         # Resolve references in the manifest
         resolved_manifest = ManifestReferenceResolver().preprocess_manifest(manifest)
+        # The check component's `config_overrides` holds connector config values rather than
+        # components, so it is read here: references are resolved, so a `check` behind a `$ref` is
+        # found, but parameters have not been propagated yet, so the values are still pristine.
+        self._check_config_overrides = self._extract_check_config_overrides(resolved_manifest)
         # Propagate types and parameters throughout the manifest
         propagated_manifest = ManifestComponentTransformer().propagate_types_and_parameters(
             "", resolved_manifest, {}
         )
 
         return propagated_manifest
 
+    @staticmethod
+    def _extract_check_config_overrides(
+        resolved_manifest: Mapping[str, Any],
+    ) -> Optional[Mapping[str, Any]]:
+        """Take the check component's `config_overrides` before parameter propagation mangles it.
+
+        `ManifestComponentTransformer` injects `$parameters` and the parameters themselves into every
+        nested mapping that carries a truthy `type` key, because everywhere else in a manifest such a
+        mapping is a component. `config_overrides` values come from the connector's own spec, where
+        `type` is an ordinary field name - so an override like `credentials: {type: oauth, ...}` would
+        otherwise reach the check with stray keys in it.
+        """
+        check = resolved_manifest.get("check")
+        if not isinstance(check, Mapping):
+            return None
+        config_overrides = check.get("config_overrides")
+        if not isinstance(config_overrides, Mapping):
+            return None
+        # Deep copy so the propagation that follows, which mutates in place, cannot reach these values.
+        return deepcopy(dict(config_overrides))
+
     def _fix_source_type(self, manifest: Dict[str, Any]) -> Dict[str, Any]:
         """
         Fix the source type in the manifest. This is necessary because the source type is not always set in the manifest.
@@ -413,7 +445,7 @@ def streams(self, config: Mapping[str, Any]) -> List[AbstractStream]:  # type: i
         """
 
         if self._spec_component:
-            self._spec_component.validate_config(self._config)
+            self._spec_component.validate_config(self._config_for_validation)
 
         api_budget_model = self._source_config.get("api_budget")
         if api_budget_model:
@@ -607,11 +639,167 @@ def check(self, logger: logging.Logger, config: Mapping[str, Any]) -> AirbyteCon
                 f"Expected to generate a ConnectionChecker component, but received {connection_checker.__class__}"
             )
 
-        check_succeeded, error = connection_checker.check_connection(self, logger, self._config)
+        with self._config_overridden_for_check(self._check_config_overrides):
+            check_succeeded, error = connection_checker.check_connection(self, logger, self._config)
         if not check_succeeded:
             return AirbyteConnectionStatus(status=Status.FAILED, message=repr(error))
         return AirbyteConnectionStatus(status=Status.SUCCEEDED)
 
+    @contextmanager
+    def _config_overridden_for_check(
+        self, config_overrides: Optional[Mapping[str, Any]]
+    ) -> Iterator[None]:
+        """Overlay the check component's `config_overrides` onto the config for the duration of a check.
+
+        A check and a sync legitimately want different behaviour from the same manifest: a check is
+        interactive and should fail fast with an actionable message, while a sync can afford to wait out
+        a rate limit window. Before this existed, expressing that difference required a Python
+        `check_connection` override that built a second component tree from a modified config, which is
+        not available to a manifest-only connector.
+
+        Overlaying here is enough to reach every component the checker builds, because `streams()`
+        interpolates from `self._config` and ignores its own `config` argument.
+
+        Semantics, all deliberate:
+          - Values are applied verbatim. They are not interpolated, so a value containing `{{ }}` reaches
+            components as that literal string.
+          - The merge is one level deep. Overriding a key whose value is an object replaces that object
+            rather than merging into it, which keeps removing a nested key expressible.
+          - The overlay happens long after `_migrate_and_transform_config`, so an override is neither
+            normalised itself nor propagated to fields derived from it by a `ConfigTransformation`.
+          - `config_validations` run against `self._config_for_validation`, so an override cannot fail a
+            validation written for the config as supplied.
+
+        Two consequences of the copy being shallow and of `self._config` being shared state:
+          - Rebinding `self._config` is not thread safe. Acceptable only because `check` is one command
+            per process.
+          - The copy is shallow, so every nested object is the *same* object as in the config the source
+            was constructed with. A component that writes into a nested path - `dpath.new(config,
+            ("credentials", "access_token"), ...)` - therefore writes through to that shared dict, and the
+            restore does not undo it, because the restore only rebinds the top level. Only a write to a
+            top-level key is discarded.
+
+        The write-through above is why a manifest that declares a `refresh_token_updater` is rejected
+        outright rather than documented: see `_raise_if_config_is_persisted`.
+        """
+        if not config_overrides:
+            yield
+            return
+
+        self._raise_on_reserved_override_keys(config_overrides)
+        self._raise_if_config_is_persisted(config_overrides)
+        self._warn_on_unknown_override_keys(config_overrides)
+        # Keys only. An override may name a secret field, so values must not reach the logs.
+        self.logger.info(
+            f"Overriding config keys for the check operation: {', '.join(sorted(config_overrides))}"
+        )
+
+        unmodified_config = self._config
+        self._config = {**self._config, **config_overrides}
+        try:
+            yield
+        finally:
+            self._config = unmodified_config
+
+    @staticmethod
+    def _raise_on_reserved_override_keys(config_overrides: Mapping[str, Any]) -> None:
+        """Refuse to overlay keys in the platform's reserved `__airbyte` namespace.
+
+        These are not connector config, they are the platform's channel into it: `CheckStream` reads
+        `__airbyte_check_stream_names` out of the very config this overlay writes to, so an override there
+        would change which streams a check tests. A manifest wanting that has `stream_names` for it. The
+        prefix is refused wholesale so future internal keys are covered too.
+        """
+        reserved = sorted(key for key in config_overrides if key.startswith("__airbyte"))
+        if reserved:
+            raise ValueError(
+                f"`config_overrides` may not set the reserved key(s) {reserved}. Keys prefixed with "
+                "`__airbyte` belong to the platform, not to the connector's spec."
+            )
+
+    def _raise_if_config_is_persisted(self, config_overrides: Mapping[str, Any]) -> None:
+        """Reject `config_overrides` on a manifest whose authenticator writes the config back.
+
+        A `refresh_token_updater` turns an `OAuthAuthenticator` into a
+        `DeclarativeSingleUseRefreshTokenOauth2Authenticator`, which is handed the config the component
+        tree was built with - during a check, the overlay. When it refreshes, `_emit_control_message`
+        prints that *entire* dict as a `CONNECTOR_CONFIG` control message, and the platform persists it.
+        So a check-only override would become the connection's saved config and apply to every later sync.
+
+        Restoring `self._config` afterwards cannot help: the control message is already on stdout. And no
+        amount of threading the config through properly would help either, because the hazard is inherent
+        to handing an overridden config to something whose job is to write the config back. Until the
+        emitter is fixed to emit the config it was given plus only the token fields it owns, refusing the
+        combination is the honest answer.
+        """
+        if not self._manifest_writes_back_config(self._source_config):
+            return
+        raise ValueError(
+            "`config_overrides` cannot be used by a manifest that declares a `refresh_token_updater`. "
+            "A token refresh during `check` emits the whole config it was handed as a CONNECTOR_CONFIG "
+            "control message, which the platform persists, so the check-only "
+            f"override(s) {sorted(config_overrides)} would be saved as this connection's config and "
+            "applied to every later sync. Remove `config_overrides`, or drop the `refresh_token_updater`."
+        )
+
+    # Subtrees of a manifest that hold data rather than components, so a `refresh_token_updater` key
+    # found inside one is a name collision and not an authenticator. Neither key is declared on any
+    # component that could contain an authenticator: `spec` exists only at the top level of a manifest,
+    # and `config_overrides` only on `CheckStream` and `CheckDynamicStream`.
+    _NON_COMPONENT_MANIFEST_KEYS = frozenset({"spec", "config_overrides"})
+
+    @staticmethod
+    def _manifest_writes_back_config(definition: Any) -> bool:
+        """Whether any component in the manifest emits the connector config back to the platform.
+
+        The scan is deliberately coarse - it looks for the key anywhere rather than only under a
+        recognised authenticator - because it runs on the raw manifest, before references are resolved.
+        An authenticator reached through a `$ref` is only found because the walk also visits
+        `definitions`, where it lives under its own key rather than under a requester. The two subtrees
+        in `_NON_COMPONENT_MANIFEST_KEYS` are the exception: they hold config values, so a matching key
+        there means a connector whose spec happens to declare a field by that name, not a token refresh.
+
+        Tested with `is not None` rather than for truthiness, to match the factory. Every field of
+        `RefreshTokenUpdater` has a default, so `refresh_token_updater: {}` is a valid way to take all of
+        them - and it builds a single-use authenticator just like a populated one, because the factory's
+        `if model.refresh_token_updater:` sees a model instance, which is always truthy. A truthiness test
+        here would see an empty dict and let that manifest through.
+        """
+        if isinstance(definition, Mapping):
+            if definition.get("refresh_token_updater") is not None:
+                return True
+            return any(
+                ConcurrentDeclarativeSource._manifest_writes_back_config(value)
+                for key, value in definition.items()
+                if key not in ConcurrentDeclarativeSource._NON_COMPONENT_MANIFEST_KEYS
+            )
+        if isinstance(definition, list):
+            return any(
+                ConcurrentDeclarativeSource._manifest_writes_back_config(item)
+                for item in definition
+            )
+        return False
+
+    def _warn_on_unknown_override_keys(self, config_overrides: Mapping[str, Any]) -> None:
+        """Warn about override keys the spec does not declare.
+
+        The overlay is the one part of the config nothing validates - the entrypoint validates what the
+        user supplied, and `config_validations` deliberately run against `self._config_for_validation`. So
+        a typo or a since-renamed field is a silent no-op that surfaces much later as "why does check no
+        longer fail fast".
+        """
+        if not self._spec_component:
+            return
+        declared = self._spec_component.connection_specification.get("properties")
+        if not isinstance(declared, Mapping):
+            return
+        unknown = sorted(key for key in config_overrides if key not in declared)
+        if unknown:
+            self.logger.warning(
+                f"Check-only config override(s) {unknown} are not declared in the connector spec, so "
+                "they will have no effect on any component that reads the config by field name."
+            )
+
     @property
     def dynamic_streams(self) -> List[Dict[str, Any]]:
         return self._dynamic_stream_configs(

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -549,6 +549,14 @@ definitions:
         type: array
         items:
           "$ref": "#/definitions/DynamicStreamCheckConfig"
+      config_overrides:
+        title: Config Overrides
+        description: Values overlaid onto the connector config for the duration of the check operation only. Use this when a check must behave differently from a sync - for example a shorter rate limit wait budget, so that check fails fast with a clear message instead of sleeping until the quota resets. Keys should be fields declared in the connector's spec. Values are used as-is. They are not interpolated, they replace a nested object rather than deep-merging into it, and they are applied after config migrations and transformations have run, so a field derived from an overridden field is not recomputed. Two combinations are rejected outright. Keys prefixed with `__airbyte` belong to the platform rather than to the connector's spec. And a manifest that declares a `refresh_token_updater` cannot use this field at all, because a token refresh during check emits the whole config it was handed as a CONNECTOR_CONFIG control message, which the platform persists - so a check-only override would become the connection's saved config and apply to every later sync.
+        type: object
+        additionalProperties: true
+        examples:
+          - max_waiting_time: 0
+          - page_size: 1
   DynamicStreamCheckConfig:
     type: object
     required:
@@ -587,6 +595,14 @@ definitions:
         description: Enables stream check availability. This field is automatically set by the CDK.
         type: boolean
         default: true
+      config_overrides:
+        title: Config Overrides
+        description: Values overlaid onto the connector config for the duration of the check operation only. Use this when a check must behave differently from a sync - for example a shorter rate limit wait budget, so that check fails fast with a clear message instead of sleeping until the quota resets. Keys should be fields declared in the connector's spec. Values are used as-is. They are not interpolated, they replace a nested object rather than deep-merging into it, and they are applied after config migrations and transformations have run, so a field derived from an overridden field is not recomputed. Two combinations are rejected outright. Keys prefixed with `__airbyte` belong to the platform rather than to the connector's spec. And a manifest that declares a `refresh_token_updater` cannot use this field at all, because a token refresh during check emits the whole config it was handed as a CONNECTOR_CONFIG control message, which the platform persists - so a check-only override would become the connection's saved config and apply to every later sync.
+        type: object
+        additionalProperties: true
+        examples:
+          - max_waiting_time: 0
+          - page_size: 1
   CompositeErrorHandler:
     title: Composite Error Handler
     description: Error handler that sequentially iterates over a list of error handlers.

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -77,6 +77,12 @@ class CheckDynamicStream(BaseModel):
         description="Enables stream check availability. This field is automatically set by the CDK.",
         title="Use Check Availability",
     )
+    config_overrides: Optional[Dict[str, Any]] = Field(
+        None,
+        description="Values overlaid onto the connector config for the duration of the check operation only. Use this when a check must behave differently from a sync - for example a shorter rate limit wait budget, so that check fails fast with a clear message instead of sleeping until the quota resets. Keys should be fields declared in the connector's spec. Values are used as-is. They are not interpolated, they replace a nested object rather than deep-merging into it, and they are applied after config migrations and transformations have run, so a field derived from an overridden field is not recomputed. Two combinations are rejected outright. Keys prefixed with `__airbyte` belong to the platform rather than to the connector's spec. And a manifest that declares a `refresh_token_updater` cannot use this field at all, because a token refresh during check emits the whole config it was handed as a CONNECTOR_CONFIG control message, which the platform persists - so a check-only override would become the connection's saved config and apply to every later sync.",
+        examples=[{"max_waiting_time": 0}, {"page_size": 1}],
+        title="Config Overrides",
+    )
 
 
 class ConcurrencyLevel(BaseModel):
@@ -1781,6 +1787,12 @@ class CheckStream(BaseModel):
         title="Stream Names",
     )
     dynamic_streams_check_configs: Optional[List[DynamicStreamCheckConfig]] = None
+    config_overrides: Optional[Dict[str, Any]] = Field(
+        None,
+        description="Values overlaid onto the connector config for the duration of the check operation only. Use this when a check must behave differently from a sync - for example a shorter rate limit wait budget, so that check fails fast with a clear message instead of sleeping until the quota resets. Keys should be fields declared in the connector's spec. Values are used as-is. They are not interpolated, they replace a nested object rather than deep-merging into it, and they are applied after config migrations and transformations have run, so a field derived from an overridden field is not recomputed. Two combinations are rejected outright. Keys prefixed with `__airbyte` belong to the platform rather than to the connector's spec. And a manifest that declares a `refresh_token_updater` cannot use this field at all, because a token refresh during check emits the whole config it was handed as a CONNECTOR_CONFIG control message, which the platform persists - so a check-only override would become the connection's saved config and apply to every later sync.",
+        examples=[{"max_waiting_time": 0}, {"page_size": 1}],
+        title="Config Overrides",
+    )
 
 
 class IncrementingCountCursor(BaseModel):