diff --git a/airbyte_cdk/connector_builder/main.py b/airbyte_cdk/connector_builder/main.py
index 06e0e3d09d..29b6fb4690 100644
--- a/airbyte_cdk/connector_builder/main.py
+++ b/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())
diff --git a/airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py b/airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py
index dea33879ac..bd189470a8 100644
--- a/airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py
+++ b/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")
diff --git a/airbyte_cdk/manifest_server/routers/manifest.py b/airbyte_cdk/manifest_server/routers/manifest.py
index ec27ca816a..0a5f0df723 100644
--- a/airbyte_cdk/manifest_server/routers/manifest.py
+++ b/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)}")
diff --git a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py
index 52f83a644d..6784816216 100644
--- a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py
+++ b/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,
diff --git a/airbyte_cdk/sources/declarative/manifest_validation_error.py b/airbyte_cdk/sources/declarative/manifest_validation_error.py
new file mode 100644
index 0000000000..35584bf64e
--- /dev/null
+++ b/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
diff --git a/unit_tests/manifest_server/routers/test_manifest.py b/unit_tests/manifest_server/routers/test_manifest.py
index e488380cb4..3acad4fd11 100644
--- a/unit_tests/manifest_server/routers/test_manifest.py
+++ b/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
diff --git a/unit_tests/sources/declarative/test_manifest_validation_error.py b/unit_tests/sources/declarative/test_manifest_validation_error.py
new file mode 100644
index 0000000000..684eb620f1
--- /dev/null
+++ b/unit_tests/sources/declarative/test_manifest_validation_error.py
@@ -0,0 +1,213 @@
+#
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+#
+
+from copy import deepcopy
+
+import pytest
+from jsonschema.exceptions import ValidationError
+
+from airbyte_cdk.legacy.sources.declarative.manifest_declarative_source import (
+    ManifestDeclarativeSource,
+)
+from airbyte_cdk.sources.declarative.concurrent_declarative_source import (
+    ConcurrentDeclarativeSource,
+)
+from airbyte_cdk.sources.declarative.manifest_validation_error import (
+    format_manifest_validation_error,
+)
+
+_BASE_MANIFEST = {
+    "version": "6.0.0",
+    "type": "DeclarativeSource",
+    "check": {"type": "CheckStream", "stream_names": ["lists"]},
+    "streams": [
+        {
+            "type": "DeclarativeStream",
+            "name": "lists",
+            "primary_key": [],
+            "schema_loader": {
+                "type": "InlineSchemaLoader",
+                "schema": {
+                    "$schema": "http://json-schema.org/schema#",
+                    "type": "object",
+                    "properties": {},
+                },
+            },
+            "retriever": {
+                "type": "SimpleRetriever",
+                "record_selector": {
+                    "type": "RecordSelector",
+                    "extractor": {"type": "DpathExtractor", "field_path": ["result"]},
+                },
+                "requester": {
+                    "type": "HttpRequester",
+                    "url_base": "https://api.example.com",
+                    "path": "/lists",
+                    "http_method": "GET",
+                    "authenticator": {
+                        "type": "SessionTokenAuthenticator",
+                        "login_requester": {
+                            "type": "HttpRequester",
+                            "url_base": "https://api.example.com",
+                            "path": "/login",
+                            "http_method": "POST",
+                        },
+                        "session_token_path": ["token"],
+                        "request_authentication": {
+                            "type": "ApiKey",
+                            "inject_into": {
+                                "type": "RequestOption",
+                                "field_name": "X",
+                                "inject_into": "header",
+                            },
+                        },
+                    },
+                },
+            },
+        }
+    ],
+}
+
+
+def _valid_manifest() -> dict:
+    return deepcopy(_BASE_MANIFEST)
+
+
+def _request_authentication(manifest: dict) -> dict:
+    return manifest["streams"][0]["retriever"]["requester"]["authenticator"][
+        "request_authentication"
+    ]
+
+
+def test_valid_manifest_constructs_without_error() -> None:
+    ConcurrentDeclarativeSource(
+        source_config=_valid_manifest(), config={}, catalog=None, state=None
+    )
+
+
+def test_invalid_authenticator_type_names_path_and_allowed_values() -> None:
+    manifest = _valid_manifest()
+    _request_authentication(manifest)["type"] = "ApiKeyAuthenticator"
+
+    with pytest.raises(ValidationError) as exc_info:
+        ConcurrentDeclarativeSource(source_config=manifest, config={}, catalog=None, state=None)
+
+    expected = (
+        "Manifest field 'streams[0].retriever.requester.authenticator."
+        "request_authentication.type' is invalid: value 'ApiKeyAuthenticator' "
+        "is not one of ['ApiKey', 'Bearer']."
+    )
+    assert str(exc_info.value) == expected
+    assert exc_info.value.message == str(exc_info.value)
+    assert "declarative_component_schema" not in str(exc_info.value)
+    assert isinstance(exc_info.value.__cause__, ValidationError)
+    assert list(exc_info.value.__cause__.absolute_path)[-1] == "request_authentication"
+
+
+def test_missing_streams_and_dynamic_streams_reports_required_properties() -> None:
+    manifest = _valid_manifest()
+    del manifest["streams"]
+
+    with pytest.raises(ValidationError) as exc_info:
+        ConcurrentDeclarativeSource(source_config=manifest, config={}, catalog=None, state=None)
+
+    assert str(exc_info.value) == (
+        "Manifest field '<root>' is invalid: one of the following properties "
+        "is required: 'streams', 'dynamic_streams'."
+    )
+
+
+def test_invalid_http_method_names_allowed_values() -> None:
+    manifest = _valid_manifest()
+    manifest["streams"][0]["retriever"]["requester"]["http_method"] = "FETCH"
+
+    with pytest.raises(ValidationError) as exc_info:
+        ConcurrentDeclarativeSource(source_config=manifest, config={}, catalog=None, state=None)
+
+    assert str(exc_info.value) == (
+        "Manifest field 'streams[0].retriever.requester.http_method' is invalid: "
+        "value 'FETCH' is not one of ['GET', 'POST']."
+    )
+
+
+def test_invalid_nested_request_option_names_path() -> None:
+    manifest = _valid_manifest()
+    _request_authentication(manifest)["inject_into"]["inject_into"] = "nowhere"
+
+    with pytest.raises(ValidationError) as exc_info:
+        ConcurrentDeclarativeSource(source_config=manifest, config={}, catalog=None, state=None)
+
+    message = str(exc_info.value)
+    assert (
+        "Manifest field 'streams[0].retriever.requester.authenticator."
+        "request_authentication.inject_into.inject_into' is invalid: "
+    ) in message
+    assert "'nowhere' is not one of [" in message
+
+
+def test_invalid_authenticator_type_in_legacy_source() -> None:
+    manifest = _valid_manifest()
+    _request_authentication(manifest)["type"] = "ApiKeyAuthenticator"
+
+    with pytest.raises(ValidationError) as exc_info:
+        ManifestDeclarativeSource(source_config=manifest)
+
+    assert str(exc_info.value) == (
+        "Manifest field 'streams[0].retriever.requester.authenticator."
+        "request_authentication.type' is invalid: value 'ApiKeyAuthenticator' "
+        "is not one of ['ApiKey', 'Bearer']."
+    )
+
+
+def test_format_type_error_single_type() -> None:
+    error = ValidationError(
+        "5 is not of type 'string'",
+        validator="type",
+        path=["a", "b"],
+        instance=5,
+        validator_value="string",
+    )
+    assert format_manifest_validation_error(error) == (
+        "Manifest field 'a.b' is invalid: value 5 is not of type 'string'."
+    )
+
+
+def test_format_type_error_multiple_types() -> None:
+    error = ValidationError(
+        "5 is not of type 'string', 'null'",
+        validator="type",
+        path=["a"],
+        instance=5,
+        validator_value=["string", "null"],
+    )
+    assert format_manifest_validation_error(error) == (
+        "Manifest field 'a' is invalid: value 5 is not of type 'string' or 'null'."
+    )
+
+
+def test_format_long_instance_repr_is_truncated() -> None:
+    error = ValidationError(
+        "long value",
+        validator="enum",
+        path=["a"],
+        instance="x" * 200,
+        validator_value=["y"],
+    )
+    message = format_manifest_validation_error(error)
+    truncated_value = repr("x" * 200)[:97] + "..."
+    assert truncated_value in message
+    assert "x" * 100 not in message
+
+
+def test_format_unknown_validator_falls_back_to_message() -> None:
+    error = ValidationError(
+        "'x' does not match '^y$'",
+        validator="pattern",
+        path=["a"],
+        instance="x",
+        validator_value="^y$",
+    )
+    assert format_manifest_validation_error(error) == (
+        "Manifest field 'a' is invalid: 'x' does not match '^y$'."
+    )