From 0704447c982d22aaba8e64fde1e558865b3b7366 Mon Sep 17 00:00:00 2001
From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Date: Wed, 20 May 2026 22:36:06 +0000
Subject: [PATCH 1/8] feat(check-stream): support configurable stream names

---
 .../declarative/checks/check_stream.py        | 14 +++-
 .../declarative/checks/test_check_stream.py   | 76 +++++++++++++++++++
 2 files changed, 89 insertions(+), 1 deletion(-)

diff --git a/airbyte_cdk/sources/declarative/checks/check_stream.py b/airbyte_cdk/sources/declarative/checks/check_stream.py
index 540813a494..7be97fb517 100644
--- a/airbyte_cdk/sources/declarative/checks/check_stream.py
+++ b/airbyte_cdk/sources/declarative/checks/check_stream.py
@@ -13,6 +13,8 @@
 from airbyte_cdk.sources.streams.core import Stream
 from airbyte_cdk.sources.streams.http.availability_strategy import HttpAvailabilityStrategy
 
+CHECK_STREAM_NAMES_CONFIG_KEY = "airbyte_check_stream_names"
+
 
 def evaluate_availability(
     stream: Union[Stream, AbstractStream], logger: logging.Logger
@@ -78,7 +80,7 @@ def check_connection(
             return self._log_error(logger, "discovering streams", error)
 
         stream_name_to_stream = {s.name: s for s in streams}
-        for stream_name in self.stream_names:
+        for stream_name in self._get_stream_names(config):
             if stream_name not in stream_name_to_stream:
                 raise ValueError(
                     f"{stream_name} is not part of the catalog. Expected one of {list(stream_name_to_stream.keys())}."
@@ -101,6 +103,16 @@ def check_connection(
 
         return True, None
 
+    def _get_stream_names(self, config: Mapping[str, Any]) -> List[str]:
+        stream_names = config.get(CHECK_STREAM_NAMES_CONFIG_KEY)
+        if stream_names is None:
+            return self.stream_names
+        if not isinstance(stream_names, list) or not all(
+            isinstance(stream_name, str) for stream_name in stream_names
+        ):
+            raise ValueError(f"{CHECK_STREAM_NAMES_CONFIG_KEY} must be a list of strings.")
+        return stream_names
+
     def _check_stream_availability(
         self,
         stream_name_to_stream: Dict[str, Union[Stream, AbstractStream]],
diff --git a/unit_tests/sources/declarative/checks/test_check_stream.py b/unit_tests/sources/declarative/checks/test_check_stream.py
index 5821ae02e2..9e7a6f2c0e 100644
--- a/unit_tests/sources/declarative/checks/test_check_stream.py
+++ b/unit_tests/sources/declarative/checks/test_check_stream.py
@@ -78,6 +78,82 @@ def mock_read_records(responses, default_response=None, **kwargs):
     )
 
 
+def test_check_stream_names_can_be_overridden_from_config():
+    static_stream = MagicMock(spec=Stream)
+    static_stream.name = "static_stream"
+    selected_stream = MagicMock(spec=Stream)
+    selected_stream.name = "selected_stream"
+    selected_stream.read_records.return_value = iter([record])
+    selected_stream.stream_slices.return_value = iter([{}])
+    source = MagicMock()
+    source.streams.return_value = [static_stream, selected_stream]
+
+    check_stream = CheckStream(["static_stream"], parameters={})
+
+    assert check_stream.check_connection(
+        source, logger, {"airbyte_check_stream_names": ["selected_stream"]}
+    ) == (True, None)
+    static_stream.stream_slices.assert_not_called()
+
+
+def test_check_stream_names_override_accepts_empty_list():
+    stream = MagicMock(spec=Stream)
+    stream.name = "static_stream"
+    source = MagicMock()
+    source.streams.return_value = [stream]
+
+    check_stream = CheckStream(["static_stream"], parameters={})
+
+    assert check_stream.check_connection(source, logger, {"airbyte_check_stream_names": []}) == (
+        True,
+        None,
+    )
+    stream.stream_slices.assert_not_called()
+
+
+@pytest.mark.parametrize("override", ["selected_stream", [1], ["selected_stream", 1]])
+def test_check_stream_names_override_requires_list_of_strings(override):
+    stream = MagicMock(spec=Stream)
+    stream.name = "selected_stream"
+    source = MagicMock()
+    source.streams.return_value = [stream]
+
+    check_stream = CheckStream(["selected_stream"], parameters={})
+
+    with pytest.raises(ValueError, match="airbyte_check_stream_names must be a list of strings."):
+        check_stream.check_connection(source, logger, {"airbyte_check_stream_names": override})
+
+
+def test_check_stream_names_override_rejects_unknown_stream():
+    stream = MagicMock(spec=Stream)
+    stream.name = "selected_stream"
+    source = MagicMock()
+    source.streams.return_value = [stream]
+
+    check_stream = CheckStream(["selected_stream"], parameters={})
+
+    with pytest.raises(ValueError, match="unknown_stream is not part of the catalog."):
+        check_stream.check_connection(
+            source, logger, {"airbyte_check_stream_names": ["unknown_stream"]}
+        )
+
+
+def test_check_stream_names_override_returns_unavailable_stream_message():
+    stream = MagicMock(spec=Stream)
+    stream.name = "selected_stream"
+    stream.stream_slices.return_value = iter([])
+    source = MagicMock()
+    source.streams.return_value = [stream]
+
+    check_stream = CheckStream(["other_stream"], parameters={})
+
+    stream_is_available, reason = check_stream.check_connection(
+        source, logger, {"airbyte_check_stream_names": ["selected_stream"]}
+    )
+    assert not stream_is_available
+    assert "no stream slices were found, likely because the parent stream is empty" in reason
+
+
 def test_check_empty_stream():
     stream = MagicMock(spec=Stream)
     stream.name = "s1"

From 344d0db698d31c17a99a637c46cb7de9143ac871 Mon Sep 17 00:00:00 2001
From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Date: Wed, 20 May 2026 23:15:28 +0000
Subject: [PATCH 2/8] refactor(check-stream): clarify configured stream names

---
 airbyte_cdk/sources/declarative/checks/check_stream.py | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/airbyte_cdk/sources/declarative/checks/check_stream.py b/airbyte_cdk/sources/declarative/checks/check_stream.py
index 7be97fb517..5d932fcb42 100644
--- a/airbyte_cdk/sources/declarative/checks/check_stream.py
+++ b/airbyte_cdk/sources/declarative/checks/check_stream.py
@@ -104,14 +104,14 @@ def check_connection(
         return True, None
 
     def _get_stream_names(self, config: Mapping[str, Any]) -> List[str]:
-        stream_names = config.get(CHECK_STREAM_NAMES_CONFIG_KEY)
-        if stream_names is None:
+        configured_stream_names = config.get(CHECK_STREAM_NAMES_CONFIG_KEY)
+        if configured_stream_names is None:
             return self.stream_names
-        if not isinstance(stream_names, list) or not all(
-            isinstance(stream_name, str) for stream_name in stream_names
+        if not isinstance(configured_stream_names, list) or not all(
+            isinstance(stream_name, str) for stream_name in configured_stream_names
         ):
             raise ValueError(f"{CHECK_STREAM_NAMES_CONFIG_KEY} must be a list of strings.")
-        return stream_names
+        return configured_stream_names
 
     def _check_stream_availability(
         self,

From 54db4058a63ce8053f50ec080cb75ae1a3025054 Mon Sep 17 00:00:00 2001
From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Date: Thu, 21 May 2026 17:33:33 +0000
Subject: [PATCH 3/8] test(check-stream): cover override edge cases

---
 .../declarative/checks/check_stream.py        |  4 +-
 .../declarative/checks/test_check_stream.py   | 69 +++++++++++++++++++
 2 files changed, 72 insertions(+), 1 deletion(-)

diff --git a/airbyte_cdk/sources/declarative/checks/check_stream.py b/airbyte_cdk/sources/declarative/checks/check_stream.py
index 5d932fcb42..6e3f6f7a95 100644
--- a/airbyte_cdk/sources/declarative/checks/check_stream.py
+++ b/airbyte_cdk/sources/declarative/checks/check_stream.py
@@ -72,6 +72,7 @@ def check_connection(
         config: Mapping[str, Any],
     ) -> Tuple[bool, Any]:
         """Checks the connection to the source and its streams."""
+        stream_names = self._get_stream_names(config)
         try:
             streams: List[Union[Stream, AbstractStream]] = source.streams(config=config)  # type: ignore  # this is a migration step and we expect the declarative CDK to migrate off of ConnectionChecker
             if not streams:
@@ -80,7 +81,7 @@ def check_connection(
             return self._log_error(logger, "discovering streams", error)
 
         stream_name_to_stream = {s.name: s for s in streams}
-        for stream_name in self._get_stream_names(config):
+        for stream_name in stream_names:
             if stream_name not in stream_name_to_stream:
                 raise ValueError(
                     f"{stream_name} is not part of the catalog. Expected one of {list(stream_name_to_stream.keys())}."
@@ -111,6 +112,7 @@ def _get_stream_names(self, config: Mapping[str, Any]) -> List[str]:
             isinstance(stream_name, str) for stream_name in configured_stream_names
         ):
             raise ValueError(f"{CHECK_STREAM_NAMES_CONFIG_KEY} must be a list of strings.")
+        # An empty override intentionally skips static stream checks; dynamic stream checks still run when configured.
         return configured_stream_names
 
     def _check_stream_availability(
diff --git a/unit_tests/sources/declarative/checks/test_check_stream.py b/unit_tests/sources/declarative/checks/test_check_stream.py
index 9e7a6f2c0e..cb45d05448 100644
--- a/unit_tests/sources/declarative/checks/test_check_stream.py
+++ b/unit_tests/sources/declarative/checks/test_check_stream.py
@@ -81,8 +81,10 @@ def mock_read_records(responses, default_response=None, **kwargs):
 def test_check_stream_names_can_be_overridden_from_config():
     static_stream = MagicMock(spec=Stream)
     static_stream.name = "static_stream"
+    static_stream.availability_strategy = None
     selected_stream = MagicMock(spec=Stream)
     selected_stream.name = "selected_stream"
+    selected_stream.availability_strategy = None
     selected_stream.read_records.return_value = iter([record])
     selected_stream.stream_slices.return_value = iter([{}])
     source = MagicMock()
@@ -99,6 +101,7 @@ def test_check_stream_names_can_be_overridden_from_config():
 def test_check_stream_names_override_accepts_empty_list():
     stream = MagicMock(spec=Stream)
     stream.name = "static_stream"
+    stream.availability_strategy = None
     source = MagicMock()
     source.streams.return_value = [stream]
 
@@ -115,6 +118,7 @@ def test_check_stream_names_override_accepts_empty_list():
 def test_check_stream_names_override_requires_list_of_strings(override):
     stream = MagicMock(spec=Stream)
     stream.name = "selected_stream"
+    stream.availability_strategy = None
     source = MagicMock()
     source.streams.return_value = [stream]
 
@@ -127,6 +131,7 @@ def test_check_stream_names_override_requires_list_of_strings(override):
 def test_check_stream_names_override_rejects_unknown_stream():
     stream = MagicMock(spec=Stream)
     stream.name = "selected_stream"
+    stream.availability_strategy = None
     source = MagicMock()
     source.streams.return_value = [stream]
 
@@ -141,6 +146,7 @@ def test_check_stream_names_override_rejects_unknown_stream():
 def test_check_stream_names_override_returns_unavailable_stream_message():
     stream = MagicMock(spec=Stream)
     stream.name = "selected_stream"
+    stream.availability_strategy = None
     stream.stream_slices.return_value = iter([])
     source = MagicMock()
     source.streams.return_value = [stream]
@@ -154,6 +160,18 @@ def test_check_stream_names_override_returns_unavailable_stream_message():
     assert "no stream slices were found, likely because the parent stream is empty" in reason
 
 
+def test_check_stream_names_override_validates_before_stream_discovery():
+    source = MagicMock()
+    check_stream = CheckStream(["selected_stream"], parameters={})
+
+    with pytest.raises(ValueError, match="airbyte_check_stream_names must be a list of strings."):
+        check_stream.check_connection(
+            source, logger, {"airbyte_check_stream_names": "selected_stream"}
+        )
+
+    source.streams.assert_not_called()
+
+
 def test_check_empty_stream():
     stream = MagicMock(spec=Stream)
     stream.name = "s1"
@@ -776,6 +794,57 @@ def test_check_stream1(
             assert connection_status.status == expected_result
 
 
+def test_check_empty_static_stream_override_still_checks_dynamic_streams():
+    manifest = {
+        **deepcopy(_MANIFEST_WITHOUT_CHECK_COMPONENT),
+        **{
+            "check": {
+                "type": "CheckStream",
+                "stream_names": ["static_stream"],
+                "dynamic_streams_check_configs": [
+                    {
+                        "type": "DynamicStreamCheckConfig",
+                        "dynamic_stream_name": "http_dynamic_stream",
+                    },
+                ],
+            }
+        },
+    }
+    check_config = {**_CONFIG, "airbyte_check_stream_names": []}
+
+    with HttpMocker() as http_mocker:
+        static_stream_request = HttpRequest(url="https://api.test.com/static")
+        static_stream_response = HttpResponse(body=json.dumps([]), status_code=500)
+        http_mocker.get(static_stream_request, static_stream_response)
+
+        items_request = HttpRequest(url="https://api.test.com/items")
+        items_response = HttpResponse(
+            body=json.dumps([{"id": 1, "name": "item_1"}, {"id": 2, "name": "item_2"}])
+        )
+        http_mocker.get(items_request, items_response)
+
+        item_request_1 = HttpRequest(url="https://api.test.com/items/1")
+        item_response = HttpResponse(body=json.dumps([]), status_code=200)
+        http_mocker.get(item_request_1, item_response)
+
+        item_request_2 = HttpRequest(url="https://api.test.com/items/2")
+        item_response = HttpResponse(body=json.dumps([]), status_code=200)
+        http_mocker.get(item_request_2, item_response)
+
+        source = ConcurrentDeclarativeSource(
+            source_config=manifest,
+            config=check_config,
+            catalog=None,
+            state=None,
+        )
+
+        connection_status = source.check(logger, check_config)
+
+        http_mocker.assert_number_of_calls(static_stream_request, 0)
+        http_mocker.assert_number_of_calls(item_request_2, 1)
+        assert connection_status.status == Status.SUCCEEDED
+
+
 def test_check_stream_missing_fields():
     """Test if ValueError is raised when dynamic_streams_check_configs is missing required fields."""
     manifest = {

From cfa68bd3195ca836f79ed3355a07d1241f92ca0f Mon Sep 17 00:00:00 2001
From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Date: Thu, 21 May 2026 18:36:20 +0000
Subject: [PATCH 4/8] test: reject null check stream overrides

---
 airbyte_cdk/sources/declarative/checks/check_stream.py     | 4 ++--
 unit_tests/sources/declarative/checks/test_check_stream.py | 2 +-
 2 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/airbyte_cdk/sources/declarative/checks/check_stream.py b/airbyte_cdk/sources/declarative/checks/check_stream.py
index 6e3f6f7a95..3c8c292225 100644
--- a/airbyte_cdk/sources/declarative/checks/check_stream.py
+++ b/airbyte_cdk/sources/declarative/checks/check_stream.py
@@ -105,9 +105,9 @@ def check_connection(
         return True, None
 
     def _get_stream_names(self, config: Mapping[str, Any]) -> List[str]:
-        configured_stream_names = config.get(CHECK_STREAM_NAMES_CONFIG_KEY)
-        if configured_stream_names is None:
+        if CHECK_STREAM_NAMES_CONFIG_KEY not in config:
             return self.stream_names
+        configured_stream_names = config[CHECK_STREAM_NAMES_CONFIG_KEY]
         if not isinstance(configured_stream_names, list) or not all(
             isinstance(stream_name, str) for stream_name in configured_stream_names
         ):
diff --git a/unit_tests/sources/declarative/checks/test_check_stream.py b/unit_tests/sources/declarative/checks/test_check_stream.py
index cb45d05448..1cb3cad57b 100644
--- a/unit_tests/sources/declarative/checks/test_check_stream.py
+++ b/unit_tests/sources/declarative/checks/test_check_stream.py
@@ -114,7 +114,7 @@ def test_check_stream_names_override_accepts_empty_list():
     stream.stream_slices.assert_not_called()
 
 
-@pytest.mark.parametrize("override", ["selected_stream", [1], ["selected_stream", 1]])
+@pytest.mark.parametrize("override", ["selected_stream", [1], ["selected_stream", 1], None])
 def test_check_stream_names_override_requires_list_of_strings(override):
     stream = MagicMock(spec=Stream)
     stream.name = "selected_stream"

From 1e15a0cb5574edfa803599912de548a62e92eb51 Mon Sep 17 00:00:00 2001
From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Date: Thu, 21 May 2026 19:03:56 +0000
Subject: [PATCH 5/8] fix(check-stream): prefix internal config override key

---
 .../sources/declarative/checks/check_stream.py |  2 +-
 .../declarative/checks/test_check_stream.py    | 18 +++++++++---------
 2 files changed, 10 insertions(+), 10 deletions(-)

diff --git a/airbyte_cdk/sources/declarative/checks/check_stream.py b/airbyte_cdk/sources/declarative/checks/check_stream.py
index 3c8c292225..afaf05a750 100644
--- a/airbyte_cdk/sources/declarative/checks/check_stream.py
+++ b/airbyte_cdk/sources/declarative/checks/check_stream.py
@@ -13,7 +13,7 @@
 from airbyte_cdk.sources.streams.core import Stream
 from airbyte_cdk.sources.streams.http.availability_strategy import HttpAvailabilityStrategy
 
-CHECK_STREAM_NAMES_CONFIG_KEY = "airbyte_check_stream_names"
+CHECK_STREAM_NAMES_CONFIG_KEY = "__airbyte_check_stream_names"
 
 
 def evaluate_availability(
diff --git a/unit_tests/sources/declarative/checks/test_check_stream.py b/unit_tests/sources/declarative/checks/test_check_stream.py
index 1cb3cad57b..fd30ecb560 100644
--- a/unit_tests/sources/declarative/checks/test_check_stream.py
+++ b/unit_tests/sources/declarative/checks/test_check_stream.py
@@ -93,7 +93,7 @@ def test_check_stream_names_can_be_overridden_from_config():
     check_stream = CheckStream(["static_stream"], parameters={})
 
     assert check_stream.check_connection(
-        source, logger, {"airbyte_check_stream_names": ["selected_stream"]}
+        source, logger, {"__airbyte_check_stream_names": ["selected_stream"]}
     ) == (True, None)
     static_stream.stream_slices.assert_not_called()
 
@@ -107,7 +107,7 @@ def test_check_stream_names_override_accepts_empty_list():
 
     check_stream = CheckStream(["static_stream"], parameters={})
 
-    assert check_stream.check_connection(source, logger, {"airbyte_check_stream_names": []}) == (
+    assert check_stream.check_connection(source, logger, {"__airbyte_check_stream_names": []}) == (
         True,
         None,
     )
@@ -124,8 +124,8 @@ def test_check_stream_names_override_requires_list_of_strings(override):
 
     check_stream = CheckStream(["selected_stream"], parameters={})
 
-    with pytest.raises(ValueError, match="airbyte_check_stream_names must be a list of strings."):
-        check_stream.check_connection(source, logger, {"airbyte_check_stream_names": override})
+    with pytest.raises(ValueError, match="__airbyte_check_stream_names must be a list of strings."):
+        check_stream.check_connection(source, logger, {"__airbyte_check_stream_names": override})
 
 
 def test_check_stream_names_override_rejects_unknown_stream():
@@ -139,7 +139,7 @@ def test_check_stream_names_override_rejects_unknown_stream():
 
     with pytest.raises(ValueError, match="unknown_stream is not part of the catalog."):
         check_stream.check_connection(
-            source, logger, {"airbyte_check_stream_names": ["unknown_stream"]}
+            source, logger, {"__airbyte_check_stream_names": ["unknown_stream"]}
         )
 
 
@@ -154,7 +154,7 @@ def test_check_stream_names_override_returns_unavailable_stream_message():
     check_stream = CheckStream(["other_stream"], parameters={})
 
     stream_is_available, reason = check_stream.check_connection(
-        source, logger, {"airbyte_check_stream_names": ["selected_stream"]}
+        source, logger, {"__airbyte_check_stream_names": ["selected_stream"]}
     )
     assert not stream_is_available
     assert "no stream slices were found, likely because the parent stream is empty" in reason
@@ -164,9 +164,9 @@ def test_check_stream_names_override_validates_before_stream_discovery():
     source = MagicMock()
     check_stream = CheckStream(["selected_stream"], parameters={})
 
-    with pytest.raises(ValueError, match="airbyte_check_stream_names must be a list of strings."):
+    with pytest.raises(ValueError, match="__airbyte_check_stream_names must be a list of strings."):
         check_stream.check_connection(
-            source, logger, {"airbyte_check_stream_names": "selected_stream"}
+            source, logger, {"__airbyte_check_stream_names": "selected_stream"}
         )
 
     source.streams.assert_not_called()
@@ -810,7 +810,7 @@ def test_check_empty_static_stream_override_still_checks_dynamic_streams():
             }
         },
     }
-    check_config = {**_CONFIG, "airbyte_check_stream_names": []}
+    check_config = {**_CONFIG, "__airbyte_check_stream_names": []}
 
     with HttpMocker() as http_mocker:
         static_stream_request = HttpRequest(url="https://api.test.com/static")

From 190f8af712081de8c6d6669ac58179dc9897e5a5 Mon Sep 17 00:00:00 2001
From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Date: Thu, 21 May 2026 20:06:29 +0000
Subject: [PATCH 6/8] Revert "fix(check-stream): prefix internal config
 override key"

This reverts commit 1e15a0cb5574edfa803599912de548a62e92eb51.
---
 .../sources/declarative/checks/check_stream.py |  2 +-
 .../declarative/checks/test_check_stream.py    | 18 +++++++++---------
 2 files changed, 10 insertions(+), 10 deletions(-)

diff --git a/airbyte_cdk/sources/declarative/checks/check_stream.py b/airbyte_cdk/sources/declarative/checks/check_stream.py
index afaf05a750..3c8c292225 100644
--- a/airbyte_cdk/sources/declarative/checks/check_stream.py
+++ b/airbyte_cdk/sources/declarative/checks/check_stream.py
@@ -13,7 +13,7 @@
 from airbyte_cdk.sources.streams.core import Stream
 from airbyte_cdk.sources.streams.http.availability_strategy import HttpAvailabilityStrategy
 
-CHECK_STREAM_NAMES_CONFIG_KEY = "__airbyte_check_stream_names"
+CHECK_STREAM_NAMES_CONFIG_KEY = "airbyte_check_stream_names"
 
 
 def evaluate_availability(
diff --git a/unit_tests/sources/declarative/checks/test_check_stream.py b/unit_tests/sources/declarative/checks/test_check_stream.py
index fd30ecb560..1cb3cad57b 100644
--- a/unit_tests/sources/declarative/checks/test_check_stream.py
+++ b/unit_tests/sources/declarative/checks/test_check_stream.py
@@ -93,7 +93,7 @@ def test_check_stream_names_can_be_overridden_from_config():
     check_stream = CheckStream(["static_stream"], parameters={})
 
     assert check_stream.check_connection(
-        source, logger, {"__airbyte_check_stream_names": ["selected_stream"]}
+        source, logger, {"airbyte_check_stream_names": ["selected_stream"]}
     ) == (True, None)
     static_stream.stream_slices.assert_not_called()
 
@@ -107,7 +107,7 @@ def test_check_stream_names_override_accepts_empty_list():
 
     check_stream = CheckStream(["static_stream"], parameters={})
 
-    assert check_stream.check_connection(source, logger, {"__airbyte_check_stream_names": []}) == (
+    assert check_stream.check_connection(source, logger, {"airbyte_check_stream_names": []}) == (
         True,
         None,
     )
@@ -124,8 +124,8 @@ def test_check_stream_names_override_requires_list_of_strings(override):
 
     check_stream = CheckStream(["selected_stream"], parameters={})
 
-    with pytest.raises(ValueError, match="__airbyte_check_stream_names must be a list of strings."):
-        check_stream.check_connection(source, logger, {"__airbyte_check_stream_names": override})
+    with pytest.raises(ValueError, match="airbyte_check_stream_names must be a list of strings."):
+        check_stream.check_connection(source, logger, {"airbyte_check_stream_names": override})
 
 
 def test_check_stream_names_override_rejects_unknown_stream():
@@ -139,7 +139,7 @@ def test_check_stream_names_override_rejects_unknown_stream():
 
     with pytest.raises(ValueError, match="unknown_stream is not part of the catalog."):
         check_stream.check_connection(
-            source, logger, {"__airbyte_check_stream_names": ["unknown_stream"]}
+            source, logger, {"airbyte_check_stream_names": ["unknown_stream"]}
         )
 
 
@@ -154,7 +154,7 @@ def test_check_stream_names_override_returns_unavailable_stream_message():
     check_stream = CheckStream(["other_stream"], parameters={})
 
     stream_is_available, reason = check_stream.check_connection(
-        source, logger, {"__airbyte_check_stream_names": ["selected_stream"]}
+        source, logger, {"airbyte_check_stream_names": ["selected_stream"]}
     )
     assert not stream_is_available
     assert "no stream slices were found, likely because the parent stream is empty" in reason
@@ -164,9 +164,9 @@ def test_check_stream_names_override_validates_before_stream_discovery():
     source = MagicMock()
     check_stream = CheckStream(["selected_stream"], parameters={})
 
-    with pytest.raises(ValueError, match="__airbyte_check_stream_names must be a list of strings."):
+    with pytest.raises(ValueError, match="airbyte_check_stream_names must be a list of strings."):
         check_stream.check_connection(
-            source, logger, {"__airbyte_check_stream_names": "selected_stream"}
+            source, logger, {"airbyte_check_stream_names": "selected_stream"}
         )
 
     source.streams.assert_not_called()
@@ -810,7 +810,7 @@ def test_check_empty_static_stream_override_still_checks_dynamic_streams():
             }
         },
     }
-    check_config = {**_CONFIG, "__airbyte_check_stream_names": []}
+    check_config = {**_CONFIG, "airbyte_check_stream_names": []}
 
     with HttpMocker() as http_mocker:
         static_stream_request = HttpRequest(url="https://api.test.com/static")

From 6ec97a4301e9daad81dbdb94c0021acd45730f20 Mon Sep 17 00:00:00 2001
From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Date: Thu, 21 May 2026 20:23:14 +0000
Subject: [PATCH 7/8] Revert "Revert "fix(check-stream): prefix internal config
 override key""

This reverts commit 190f8af712081de8c6d6669ac58179dc9897e5a5.
---
 .../sources/declarative/checks/check_stream.py |  2 +-
 .../declarative/checks/test_check_stream.py    | 18 +++++++++---------
 2 files changed, 10 insertions(+), 10 deletions(-)

diff --git a/airbyte_cdk/sources/declarative/checks/check_stream.py b/airbyte_cdk/sources/declarative/checks/check_stream.py
index 3c8c292225..afaf05a750 100644
--- a/airbyte_cdk/sources/declarative/checks/check_stream.py
+++ b/airbyte_cdk/sources/declarative/checks/check_stream.py
@@ -13,7 +13,7 @@
 from airbyte_cdk.sources.streams.core import Stream
 from airbyte_cdk.sources.streams.http.availability_strategy import HttpAvailabilityStrategy
 
-CHECK_STREAM_NAMES_CONFIG_KEY = "airbyte_check_stream_names"
+CHECK_STREAM_NAMES_CONFIG_KEY = "__airbyte_check_stream_names"
 
 
 def evaluate_availability(
diff --git a/unit_tests/sources/declarative/checks/test_check_stream.py b/unit_tests/sources/declarative/checks/test_check_stream.py
index 1cb3cad57b..fd30ecb560 100644
--- a/unit_tests/sources/declarative/checks/test_check_stream.py
+++ b/unit_tests/sources/declarative/checks/test_check_stream.py
@@ -93,7 +93,7 @@ def test_check_stream_names_can_be_overridden_from_config():
     check_stream = CheckStream(["static_stream"], parameters={})
 
     assert check_stream.check_connection(
-        source, logger, {"airbyte_check_stream_names": ["selected_stream"]}
+        source, logger, {"__airbyte_check_stream_names": ["selected_stream"]}
     ) == (True, None)
     static_stream.stream_slices.assert_not_called()
 
@@ -107,7 +107,7 @@ def test_check_stream_names_override_accepts_empty_list():
 
     check_stream = CheckStream(["static_stream"], parameters={})
 
-    assert check_stream.check_connection(source, logger, {"airbyte_check_stream_names": []}) == (
+    assert check_stream.check_connection(source, logger, {"__airbyte_check_stream_names": []}) == (
         True,
         None,
     )
@@ -124,8 +124,8 @@ def test_check_stream_names_override_requires_list_of_strings(override):
 
     check_stream = CheckStream(["selected_stream"], parameters={})
 
-    with pytest.raises(ValueError, match="airbyte_check_stream_names must be a list of strings."):
-        check_stream.check_connection(source, logger, {"airbyte_check_stream_names": override})
+    with pytest.raises(ValueError, match="__airbyte_check_stream_names must be a list of strings."):
+        check_stream.check_connection(source, logger, {"__airbyte_check_stream_names": override})
 
 
 def test_check_stream_names_override_rejects_unknown_stream():
@@ -139,7 +139,7 @@ def test_check_stream_names_override_rejects_unknown_stream():
 
     with pytest.raises(ValueError, match="unknown_stream is not part of the catalog."):
         check_stream.check_connection(
-            source, logger, {"airbyte_check_stream_names": ["unknown_stream"]}
+            source, logger, {"__airbyte_check_stream_names": ["unknown_stream"]}
         )
 
 
@@ -154,7 +154,7 @@ def test_check_stream_names_override_returns_unavailable_stream_message():
     check_stream = CheckStream(["other_stream"], parameters={})
 
     stream_is_available, reason = check_stream.check_connection(
-        source, logger, {"airbyte_check_stream_names": ["selected_stream"]}
+        source, logger, {"__airbyte_check_stream_names": ["selected_stream"]}
     )
     assert not stream_is_available
     assert "no stream slices were found, likely because the parent stream is empty" in reason
@@ -164,9 +164,9 @@ def test_check_stream_names_override_validates_before_stream_discovery():
     source = MagicMock()
     check_stream = CheckStream(["selected_stream"], parameters={})
 
-    with pytest.raises(ValueError, match="airbyte_check_stream_names must be a list of strings."):
+    with pytest.raises(ValueError, match="__airbyte_check_stream_names must be a list of strings."):
         check_stream.check_connection(
-            source, logger, {"airbyte_check_stream_names": "selected_stream"}
+            source, logger, {"__airbyte_check_stream_names": "selected_stream"}
         )
 
     source.streams.assert_not_called()
@@ -810,7 +810,7 @@ def test_check_empty_static_stream_override_still_checks_dynamic_streams():
             }
         },
     }
-    check_config = {**_CONFIG, "airbyte_check_stream_names": []}
+    check_config = {**_CONFIG, "__airbyte_check_stream_names": []}
 
     with HttpMocker() as http_mocker:
         static_stream_request = HttpRequest(url="https://api.test.com/static")

From 7e52e79485e42dc231ffe84684a4b625ea4633b2 Mon Sep 17 00:00:00 2001
From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Date: Thu, 21 May 2026 21:16:32 +0000
Subject: [PATCH 8/8] fix(check-stream): fall back on empty stream override

---
 .../sources/declarative/checks/check_stream.py     |  6 ++++--
 .../declarative/checks/test_check_stream.py        | 14 ++++++++------
 2 files changed, 12 insertions(+), 8 deletions(-)

diff --git a/airbyte_cdk/sources/declarative/checks/check_stream.py b/airbyte_cdk/sources/declarative/checks/check_stream.py
index afaf05a750..618511f8d7 100644
--- a/airbyte_cdk/sources/declarative/checks/check_stream.py
+++ b/airbyte_cdk/sources/declarative/checks/check_stream.py
@@ -105,14 +105,16 @@ def check_connection(
         return True, None
 
     def _get_stream_names(self, config: Mapping[str, Any]) -> List[str]:
-        if CHECK_STREAM_NAMES_CONFIG_KEY not in config:
+        if (
+            CHECK_STREAM_NAMES_CONFIG_KEY not in config
+            or config[CHECK_STREAM_NAMES_CONFIG_KEY] == []
+        ):
             return self.stream_names
         configured_stream_names = config[CHECK_STREAM_NAMES_CONFIG_KEY]
         if not isinstance(configured_stream_names, list) or not all(
             isinstance(stream_name, str) for stream_name in configured_stream_names
         ):
             raise ValueError(f"{CHECK_STREAM_NAMES_CONFIG_KEY} must be a list of strings.")
-        # An empty override intentionally skips static stream checks; dynamic stream checks still run when configured.
         return configured_stream_names
 
     def _check_stream_availability(
diff --git a/unit_tests/sources/declarative/checks/test_check_stream.py b/unit_tests/sources/declarative/checks/test_check_stream.py
index fd30ecb560..8bc9571d1a 100644
--- a/unit_tests/sources/declarative/checks/test_check_stream.py
+++ b/unit_tests/sources/declarative/checks/test_check_stream.py
@@ -98,10 +98,12 @@ def test_check_stream_names_can_be_overridden_from_config():
     static_stream.stream_slices.assert_not_called()
 
 
-def test_check_stream_names_override_accepts_empty_list():
+def test_check_stream_names_override_empty_list_falls_back_to_manifest_streams():
     stream = MagicMock(spec=Stream)
     stream.name = "static_stream"
     stream.availability_strategy = None
+    stream.read_records.return_value = iter([record])
+    stream.stream_slices.return_value = iter([{}])
     source = MagicMock()
     source.streams.return_value = [stream]
 
@@ -111,7 +113,7 @@ def test_check_stream_names_override_accepts_empty_list():
         True,
         None,
     )
-    stream.stream_slices.assert_not_called()
+    stream.stream_slices.assert_called_once()
 
 
 @pytest.mark.parametrize("override", ["selected_stream", [1], ["selected_stream", 1], None])
@@ -794,7 +796,7 @@ def test_check_stream1(
             assert connection_status.status == expected_result
 
 
-def test_check_empty_static_stream_override_still_checks_dynamic_streams():
+def test_check_empty_static_stream_override_falls_back_to_manifest_streams_and_checks_dynamic_streams():
     manifest = {
         **deepcopy(_MANIFEST_WITHOUT_CHECK_COMPONENT),
         **{
@@ -840,9 +842,9 @@ def test_check_empty_static_stream_override_still_checks_dynamic_streams():
 
         connection_status = source.check(logger, check_config)
 
-        http_mocker.assert_number_of_calls(static_stream_request, 0)
-        http_mocker.assert_number_of_calls(item_request_2, 1)
-        assert connection_status.status == Status.SUCCEEDED
+        http_mocker.assert_number_of_calls(static_stream_request, 6)
+        http_mocker.assert_number_of_calls(item_request_2, 0)
+        assert connection_status.status == Status.FAILED
 
 
 def test_check_stream_missing_fields():