fix(cloud): treat an unavailable feature probe as unknown, not disabled

airbyte/cloud/connectors.py

@@ -88,7 +88,7 @@ class _ConnectorDefinitionLike(Protocol):
     docker_repository: str
 
 
-class CloudConnector:
+class CloudConnector:  # noqa: PLR0904 - public connector management and execution API
     """A cloud connector is a deployed source or destination on Airbyte Cloud.
 
     You can use a connector object to manage the connector.
@@ -122,17 +122,29 @@ def __init__(
         self._connector_definition: _ConnectorDefinitionLike | None = None
         """The connector definition lookup result. (Cached; `None` until fetched.)"""
 
-    def _get_enabled_features(self) -> frozenset[ConnectorFeature]:
-        """Return the enabled features, resolving them through the workspace on first use."""
-        if self._enabled_features is None:
-            self._enabled_features = self.workspace._get_connector_features(self)  # noqa: SLF001
+    def get_enabled_features(
+        self, *, warnings: list[str] | None = None
+    ) -> frozenset[ConnectorFeature] | None:
+        """Resolve enabled features; `None` means the docs probe was unavailable (403/404).
 
+        Results are cached only when known. Probe failures other than 403/404 propagate.
+        """
+        if self._enabled_features is None:
+            features = self.workspace._get_connector_features(self, warnings=warnings)  # noqa: SLF001
+            if features is None:
+                return None
+            self._enabled_features = features
         return self._enabled_features
 
     @property
     def enabled_features(self) -> frozenset[ConnectorFeature]:
-        """The features enabled for this connector. Resolved on first access and cached."""
-        return self._get_enabled_features()
+        """The features enabled for this connector. Resolved on first access and cached.
+
+        An unknown result (unavailable probe) yields an empty set; use
+        `get_enabled_features()` to distinguish unknown from known-empty.
+        """
+        features = self.get_enabled_features()
+        return frozenset() if features is None else features
 
     def is_feature_enabled(self, feature: ConnectorFeature) -> bool:
         """Whether `feature` is enabled for this connector.

airbyte/cloud/workspaces.py

@@ -427,25 +427,29 @@ def is_feature_enabled(self, feature: OrganizationFeature) -> bool:
     def _get_connector_features(
         self,
         connector: cloud_connectors.CloudConnector,
-    ) -> frozenset[ConnectorFeature]:
+        *,
+        warnings: list[str] | None = None,
+    ) -> frozenset[ConnectorFeature] | None:
         """Resolve the enabled features for one connector in this workspace.
 
         A docs probe against the Context layer reports whether the connector
-        is enabled for agent access. Search indexing has not launched yet, so it is never
-        reported as enabled.
+        is enabled for agent access; an unavailable probe returns `None` since it
+        does not prove disabled access. Search indexing has not launched yet, so it
+        is never reported as enabled.
         """
         if not self._has_context_layer_api():
             return frozenset()
 
+        probe_warnings = warnings if warnings is not None else []
         if connector.connector_type == ConnectorType.DESTINATION:
             if connector.definition_id not in _SQL_PASSTHROUGH_DESTINATION_DEFINITION_IDS:
                 return frozenset()
-            if connector._context_layer_inspect(warnings=[]) is None:  # noqa: SLF001
-                return frozenset()
+            if connector._context_layer_inspect(warnings=probe_warnings) is None:  # noqa: SLF001
+                return None
             return frozenset({ConnectorFeature.DIRECT_ACCESS, ConnectorFeature.DIRECT_SQL_QUERY})
 
-        if connector._context_layer_inspect(warnings=[]) is None:  # noqa: SLF001
-            return frozenset()
+        if connector._context_layer_inspect(warnings=probe_warnings) is None:  # noqa: SLF001
+            return None
 
         return frozenset({ConnectorFeature.DIRECT_ACCESS, ConnectorFeature.DIRECT_API_QUERY})
 
@@ -1179,12 +1183,8 @@ def list_connectors(
 
         matches: list[cloud_connectors.CloudConnector] = []
         for connector in connectors:
-            if feature_filter is not None:
-                connector._enabled_features = self._get_connector_features(  # noqa: SLF001
-                    connector
-                )
-                if feature_filter not in connector._get_enabled_features():  # noqa: SLF001
-                    continue
+            if feature_filter is not None and feature_filter not in connector.enabled_features:
+                continue
             matches.append(connector)
             if limit is not None and len(matches) >= limit:
                 break

airbyte/mcp/cloud.py

@@ -1205,8 +1205,9 @@ def list_cloud_connectors(
     for connector in connectors:
         features: frozenset[ConnectorFeature] | None = None
         if probe_failure is None:
+            probe_warnings: list[str] = []
             try:
-                features = connector.enabled_features
+                features = connector.get_enabled_features(warnings=probe_warnings)
             except (AirbyteError, requests.RequestException) as error:
                 warning = f"Connector feature lookup failed; enabled features are unknown: {error}"
                 if isinstance(error, requests.RequestException) or (
@@ -1233,6 +1234,24 @@ def list_cloud_connectors(
                     if limit is not None and len(results) >= limit:
                         break
                     continue
+            if features is None:
+                results.append(
+                    CloudConnectorResult(
+                        id=connector.connector_id,
+                        connector_type=connector.connector_type.value,
+                        name=cast(str, connector.name),
+                        url=connector.connector_url,
+                        enabled_features=FEATURES_UNKNOWN,
+                        warnings=probe_warnings
+                        or [
+                            "Connector direct-access docs are unavailable; "
+                            "enabled features are unknown."
+                        ],
+                    )
+                )
+                if limit is not None and len(results) >= limit:
+                    break
+                continue
         if probe_failure is not None:
             enabled_features: list[ConnectorFeature] | FeaturesUnknown = FEATURES_UNKNOWN
             connector_warnings = [probe_failure]
@@ -1335,16 +1354,10 @@ def _describe_cloud_connector(
     )
 
     try:
-        result.enabled_features = sorted(connector.enabled_features)
-        if (
-            connector.is_feature_enabled(ConnectorFeature.DIRECT_ACCESS)
-            and connector.workspace._has_context_layer_api()  # noqa: SLF001
-        ):
-            context_layer = connector._context_layer_inspect(  # noqa: SLF001
-                warnings=warnings,
-            )
-            if context_layer is not None:
-                warnings.extend(str(warning) for warning in context_layer.warnings)
+        features = connector.get_enabled_features(warnings=warnings)
+        result.enabled_features = FEATURES_UNKNOWN if features is None else sorted(features)
+        if features and connector._context_layer_details is not None:  # noqa: SLF001
+            warnings.extend(str(warning) for warning in connector._context_layer_details.warnings)  # noqa: SLF001
     except (AirbyteError, requests.RequestException) as error:
         result.enabled_features = FEATURES_UNKNOWN
         warnings.append(f"Connector feature lookup failed; enabled features are unknown: {error}")

tests/unit_tests/test_cloud_credentials.py

@@ -2080,8 +2080,8 @@ def test_cloud_destination_features_raise_on_probe_failure(
         ),
         pytest.param(
             ConnectorFeature.SEARCH_INDEXING,
-            [],
-            [],
+            ["source-3"],
+            [mcp_cloud.FEATURES_UNKNOWN],
             id="search_indexing",
         ),
     ],

tests/unit_tests/test_cloud_describe.py

@@ -32,6 +32,7 @@
     CloudSourceInfo,
 )
 from airbyte.cloud.workspaces import CloudWorkspace
+from airbyte.mcp import cloud as cloud_mcp
 from airbyte.exceptions import (
     AirbyteCloudApiError,
     AirbyteError,
@@ -774,3 +775,136 @@ def test_is_feature_enabled_wrong_connector_kind_short_circuits(
 
     assert connector.is_feature_enabled(feature) is False
     get_features.assert_not_called()
+
+
+@pytest.mark.parametrize("status", [403, 404])
+def test_get_connector_features_unavailable_probe_returns_none(
+    monkeypatch: pytest.MonkeyPatch, status: int
+) -> None:
+    """A 403/404 docs probe means 'unknown': `None` for sources and passthrough destinations."""
+    workspace = _make_workspace(monkeypatch)
+    _patch_context_layer(monkeypatch)
+    read_docs = MagicMock(
+        side_effect=AirbyteCloudApiError(
+            status_code=status, context={"response_text": "secret"}
+        )
+    )
+    monkeypatch.setattr(agents_api_util, "read_cloud_skill_docs", read_docs)
+    source = _seed_source(workspace, "source-1", "GitHub")
+    passthrough = _seed_destination(workspace, "snowflake", SNOWFLAKE_DEFINITION_ID)
+    plain = _seed_destination(workspace, "dest-2", "other-definition")
+
+    warnings: list[str] = []
+    assert workspace._get_connector_features(source, warnings=warnings) is None  # noqa: SLF001
+    assert workspace._get_connector_features(passthrough, warnings=warnings) is None  # noqa: SLF001
+    assert workspace._get_connector_features(plain) == frozenset()  # noqa: SLF001
+    assert warnings
+    assert read_docs.call_count == 2
+
+
+@pytest.mark.parametrize("status", [403, 404])
+def test_unavailable_features_warn_without_caching_absence(
+    monkeypatch: pytest.MonkeyPatch, status: int
+) -> None:
+    """An unavailable probe warns and is not cached; a later successful probe caches."""
+    workspace = _make_workspace(monkeypatch)
+    _patch_context_layer(monkeypatch)
+    read_docs = MagicMock(
+        side_effect=[
+            AirbyteCloudApiError(
+                status_code=status, context={"response_text": "secret"}
+            ),
+            SKILL_DOCS_RESPONSE,
+        ]
+    )
+    monkeypatch.setattr(agents_api_util, "read_cloud_skill_docs", read_docs)
+    source = _seed_source(workspace, "source-1", "GitHub")
+    warnings: list[str] = []
+
+    assert source.get_enabled_features(warnings=warnings) is None
+    assert source._enabled_features is None  # noqa: SLF001
+    assert len(warnings) == 1
+    assert ConnectorFeature.DIRECT_API_QUERY in source.get_enabled_features()
+    assert read_docs.call_count == 2
+    assert source.enabled_features == frozenset({
+        ConnectorFeature.DIRECT_ACCESS,
+        ConnectorFeature.DIRECT_API_QUERY,
+    })
+    assert read_docs.call_count == 2
+
+
+def test_enabled_features_property_empty_when_probe_unavailable(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The `enabled_features` property reports an empty set when the probe is unavailable."""
+    workspace = _make_workspace(monkeypatch)
+    _patch_context_layer(monkeypatch)
+    read_docs = MagicMock(
+        side_effect=AirbyteCloudApiError(
+            status_code=404, context={"response_text": "secret"}
+        )
+    )
+    monkeypatch.setattr(agents_api_util, "read_cloud_skill_docs", read_docs)
+    source = _seed_source(workspace, "source-1", "GitHub")
+
+    assert source.get_enabled_features() is None
+    assert source.enabled_features == frozenset()
+    assert read_docs.call_count == 2  # the unavailable result is not cached
+
+
+@pytest.mark.parametrize("status", [403, 404])
+def test_describe_survives_unknown_features_and_recovers(
+    monkeypatch: pytest.MonkeyPatch, status: int
+) -> None:
+    """A 403/404 probe marks described features 'unknown' with a warning and is not cached."""
+    workspace = _make_workspace(monkeypatch)
+    _patch_context_layer(monkeypatch)
+    read_docs = MagicMock(
+        side_effect=[
+            AirbyteCloudApiError(
+                status_code=status, context={"response_text": "secret"}
+            ),
+            SKILL_DOCS_RESPONSE,
+        ]
+    )
+    monkeypatch.setattr(agents_api_util, "read_cloud_skill_docs", read_docs)
+    source = _seed_source(workspace, "source-1", "GitHub")
+    source._connector_definition = SimpleNamespace(name="GitHub")  # noqa: SLF001
+    monkeypatch.setattr(connector_docs, "build_connection_details", lambda _: [])
+
+    result = cloud_mcp._describe_cloud_connector(  # noqa: SLF001
+        source,
+        with_config=False,
+        with_replication_details=True,
+        with_direct_access_guidance=False,
+        with_data_replication_docs=False,
+    )
+
+    assert result.connector_id == "source-1"
+    assert result.enabled_features == "unknown"
+    assert result.warnings
+    assert read_docs.call_count == 1
+    assert source._enabled_features is None  # noqa: SLF001
+    assert ConnectorFeature.DIRECT_API_QUERY in source.enabled_features
+    assert read_docs.call_count == 2
+
+
+def test_malformed_probe_is_not_cached_or_silently_filtered(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A malformed probe response propagates and is not cached."""
+    workspace = _make_workspace(monkeypatch)
+    _patch_context_layer(monkeypatch)
+    source = _seed_source(workspace, "source-1", "GitHub")
+    monkeypatch.setattr(workspace, "list_sources", lambda: [source])
+    monkeypatch.setattr(workspace, "list_destinations", lambda: [])
+    read_docs = MagicMock(side_effect=[{}, SKILL_DOCS_RESPONSE])
+    monkeypatch.setattr(agents_api_util, "read_cloud_skill_docs", read_docs)
+
+    with pytest.raises(ValueError):
+        workspace.list_connectors(feature_filter=ConnectorFeature.DIRECT_ACCESS)
+    assert source._enabled_features is None  # noqa: SLF001
+    assert workspace.list_connectors(feature_filter=ConnectorFeature.DIRECT_ACCESS) == [
+        source
+    ]
+    assert read_docs.call_count == 2