fix(declarative): surface failing path and constraint in manifest validation errors

airbyte_cdk/connector_builder/main.py

@@ -7,6 +7,7 @@
 from typing import Any, List, Mapping, Optional, Tuple
 
 import orjson
+from jsonschema.exceptions import ValidationError
 
 from airbyte_cdk.connector import BaseConnector
 from airbyte_cdk.connector_builder.connector_builder_handler import (
@@ -24,6 +25,7 @@
     AirbyteStateMessage,
     ConfiguredAirbyteCatalog,
     ConfiguredAirbyteCatalogSerializer,
+    FailureType,
 )
 from airbyte_cdk.sources.declarative.concurrent_declarative_source import (
     ConcurrentDeclarativeSource,
@@ -105,8 +107,13 @@ def handle_request(args: List[str]) -> str:
     try:
         print(handle_request(sys.argv[1:]))
     except Exception as exc:
+        failure_type = (
+            FailureType.config_error
+            if isinstance(exc, ValidationError)
+            else FailureType.system_error
+        )
         error = AirbyteTracedException.from_exception(
-            exc, message=f"Error handling request: {str(exc)}"
+            exc, message=f"Error handling request: {str(exc)}", failure_type=failure_type
         )
         m = error.as_airbyte_message()
         print(orjson.dumps(AirbyteMessageSerializer.dump(m)).decode())

airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py

@@ -36,6 +36,9 @@
 from airbyte_cdk.sources.declarative.checks import COMPONENTS_CHECKER_TYPE_MAPPING
 from airbyte_cdk.sources.declarative.checks.connection_checker import ConnectionChecker
 from airbyte_cdk.sources.declarative.interpolation import InterpolatedBoolean
+from airbyte_cdk.sources.declarative.manifest_validation_error import (
+    format_manifest_validation_error,
+)
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     ConditionalStreams as ConditionalStreamsModel,
 )
@@ -450,9 +453,7 @@ def _validate_source(self) -> None:
         try:
             validate(self._source_config, self._declarative_component_schema)
         except ValidationError as e:
-            raise ValidationError(
-                "Validation against json schema defined in declarative_component_schema.yaml schema failed"
-            ) from e
+            raise ValidationError(format_manifest_validation_error(e)) from e
 
         cdk_version_str = metadata.version("airbyte_cdk")
         cdk_version = self._parse_version(cdk_version_str, "airbyte-cdk")

airbyte_cdk/manifest_server/routers/manifest.py

@@ -56,7 +56,9 @@ def safe_build_source(
             slice_limit,
         )
     except jsonschema.exceptions.ValidationError as e:
-        raise HTTPException(status_code=400, detail=f"Invalid manifest: {e.message}")
+        raise HTTPException(
+            status_code=400, detail=filter_secrets(f"Invalid manifest: {e.message}")
+        )
 
 
 router = APIRouter(
@@ -213,6 +215,9 @@ def resolve(request: ResolveRequest) -> ManifestResponse:
     try:
         source = safe_build_source(request.manifest.model_dump(), {})
         return ManifestResponse(manifest=Manifest(**source.resolved_manifest))
+    except HTTPException:
+        # Re-raise HTTPExceptions as-is (like the invalid manifest check in safe_build_source)
+        raise
     except Exception as exc:
         # Filter secrets from error message before returning to client
         sanitized_message = filter_secrets(f"Error resolving manifest: {str(exc)}")
@@ -258,6 +263,9 @@ def full_resolve(request: FullResolveRequest) -> ManifestResponse:
 
         manifest["streams"] = streams
         return ManifestResponse(manifest=Manifest(**manifest))
+    except HTTPException:
+        # Re-raise HTTPExceptions as-is (like the invalid manifest check in safe_build_source)
+        raise
     except Exception as exc:
         # Filter secrets from error message before returning to client
         sanitized_message = filter_secrets(f"Error full resolving manifest: {str(exc)}")

airbyte_cdk/sources/declarative/concurrent_declarative_source.py

@@ -50,6 +50,9 @@
 from airbyte_cdk.sources.declarative.checks.connection_checker import ConnectionChecker
 from airbyte_cdk.sources.declarative.concurrency_level import ConcurrencyLevel
 from airbyte_cdk.sources.declarative.interpolation import InterpolatedBoolean
+from airbyte_cdk.sources.declarative.manifest_validation_error import (
+    format_manifest_validation_error,
+)
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     ConcurrencyLevel as ConcurrencyLevelModel,
 )
@@ -348,9 +351,7 @@ def _validate_source(self) -> None:
         try:
             validate(self._source_config, self._declarative_component_schema)
         except ValidationError as e:
-            raise ValidationError(
-                "Validation against json schema defined in declarative_component_schema.yaml schema failed"
-            ) from e
+            raise ValidationError(format_manifest_validation_error(e)) from e
 
     def _migrate_and_transform_config(
         self,

airbyte_cdk/sources/declarative/manifest_validation_error.py

@@ -0,0 +1,106 @@
+#
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+#
+
+from collections import defaultdict
+from typing import Any, Dict, List, Optional, Sequence, Tuple, Union
+
+from jsonschema.exceptions import ValidationError
+
+_MAX_VALUE_REPR_LENGTH = 100
+_ROOT_PATH = "<root>"
+
+
+def format_manifest_validation_error(error: ValidationError) -> str:
+    """Builds a user-facing message naming the failing manifest field and the violated constraint.
+
+    `error` is expected to be the error raised by `jsonschema.validate`, i.e. the result of
+    `jsonschema.exceptions.best_match`. For `anyOf`/`oneOf` errors, the sub-errors of the branches
+    are merged so that e.g. a wrong `type` discriminator reports every allowed value instead of
+    the opaque "is not valid under any of the given schemas".
+    """
+    path, constraint = _describe(error)
+    return f"Manifest field '{path}' is invalid: {constraint}."
+
+
+def _describe(error: ValidationError) -> Tuple[str, str]:
+    base_path: List[Union[str, int]] = list(error.absolute_path)
+    if error.validator in ("anyOf", "oneOf") and error.context:
+        merged = _merge_branch_errors(error, base_path)
+        if merged:
+            return merged
+        return (
+            _format_path(base_path),
+            "the value does not match any of the allowed component definitions",
+        )
+    return _format_path(base_path), _describe_single(error)
+
+
+def _merge_branch_errors(
+    error: ValidationError, base_path: List[Union[str, int]]
+) -> Optional[Tuple[str, str]]:
+    by_path: Dict[Tuple[Union[str, int], ...], List[ValidationError]] = defaultdict(list)
+    for sub_error in error.context:
+        by_path[tuple(sub_error.relative_path)].append(sub_error)
+    shallowest = min(by_path, key=len)
+    candidates = by_path[shallowest]
+    validators = {sub_error.validator for sub_error in candidates}
+    if validators == {"enum"}:
+        allowed: List[Any] = []
+        for sub_error in candidates:
+            for value in sub_error.validator_value:
+                if value not in allowed:
+                    allowed.append(value)
+        return (
+            _format_path(base_path + list(shallowest)),
+            f"value {_short_repr(candidates[0].instance)} is not one of {_format_options(allowed)}",
+        )
+    if validators == {"required"}:
+        present = candidates[0].instance if isinstance(candidates[0].instance, dict) else {}
+        missing: List[str] = []
+        for sub_error in candidates:
+            for prop in sub_error.validator_value:
+                if prop not in present and prop not in missing:
+                    missing.append(prop)
+        return (
+            _format_path(base_path + list(shallowest)),
+            f"one of the following properties is required: {', '.join(repr(prop) for prop in missing)}",
+        )
+    return None
+
+
+def _describe_single(error: ValidationError) -> str:
+    if error.validator == "enum":
+        return f"value {_short_repr(error.instance)} is not one of {_format_options(error.validator_value)}"
+    if error.validator == "type":
+        expected = error.validator_value
+        expected_str = (
+            " or ".join(f"'{t}'" for t in expected)
+            if isinstance(expected, list)
+            else f"'{expected}'"
+        )
+        return f"value {_short_repr(error.instance)} is not of type {expected_str}"
+    return str(error.message)
+
+
+def _format_path(path: Sequence[Union[str, int]]) -> str:
+    if not path:
+        return _ROOT_PATH
+    formatted = ""
+    for part in path:
+        if isinstance(part, int):
+            formatted += f"[{part}]"
+        else:
+            formatted += f".{part}" if formatted else str(part)
+    return formatted
+
+
+def _format_options(options: Sequence[Any]) -> str:
+    return "[" + ", ".join(repr(option) for option in options) + "]"
+
+
+def _short_repr(value: Any) -> str:
+    text = repr(value)
+    if len(text) > _MAX_VALUE_REPR_LENGTH:
+        return text[: _MAX_VALUE_REPR_LENGTH - 3] + "..."
+    return text

unit_tests/manifest_server/routers/test_manifest.py

@@ -2,10 +2,13 @@
 from unittest.mock import Mock, patch
 
 import pytest
+from fastapi import HTTPException
 from fastapi.testclient import TestClient
 
 from airbyte_cdk.connector_builder.models import StreamRead as CDKStreamRead
 from airbyte_cdk.manifest_server.app import app
+from airbyte_cdk.manifest_server.routers.manifest import safe_build_source
+from airbyte_cdk.utils.airbyte_secrets_utils import update_secrets
 
 client = TestClient(app)
 
@@ -270,6 +273,27 @@ def test_resolve_invalid_manifest(self):
         response = client.post("/v1/manifest/resolve", json=request_data)
         assert response.status_code == 422  # Validation error
 
+    def test_safe_build_source_filters_secrets_from_validation_error(
+        self, sample_manifest, sample_config
+    ):
+        """Secrets used as invalid manifest values must not leak into the HTTPException detail."""
+        secret = "my-super-secret-http-method"
+        sample_manifest["streams"][0]["retriever"]["requester"]["http_method"] = secret
+
+        update_secrets([secret])
+        try:
+            with pytest.raises(HTTPException) as exc_info:
+                safe_build_source(sample_manifest, sample_config)
+        finally:
+            update_secrets([])
+
+        assert exc_info.value.status_code == 400
+        assert secret not in exc_info.value.detail
+        assert (
+            "Invalid manifest: Manifest field "
+            "'streams[0].retriever.requester.http_method'" in exc_info.value.detail
+        )
+
     def test_full_resolve_endpoint_success(self, sample_manifest, sample_config, mock_source):
         """Test successful full_resolve endpoint call."""
         # Setup mock source with dynamic streams