feat(exceptions): expose __cause__ as original_exception

airbyte/_connector_base.py

@@ -383,7 +383,7 @@ def check(self) -> None:
                 raise exc.AirbyteConnectorCheckFailedError(
                     connector_name=self.name,
                     original_exception=ex,
-                ) from None
+                ) from ex
 
     def install(self) -> None:
         """Install the connector if it is not yet installed."""
@@ -518,7 +518,7 @@ def _execute(
                 connector_name=self.name,
                 log_text=self._last_log_messages,
                 original_exception=e,
-            ) from None
+            ) from e
 
 
 __all__ = [

airbyte/_util/destination_smoke_tests.py

@@ -265,6 +265,14 @@ def _prepare_destination_config(
         destination.set_config(config, validate=False)
 
 
+def _get_opted_in_original_exception(error: BaseException) -> BaseException | None:
+    if not getattr(error, "_original_exception_opt_in", False):
+        return None
+
+    original_exception = getattr(error, "original_exception", None)
+    return original_exception if isinstance(original_exception, BaseException) else None
+
+
 def _extract_trace_error_from_log(ex: Exception) -> str | None:
     """Search the connector log file for a TRACE ERROR `internal_message`.
 
@@ -278,12 +286,13 @@ def _extract_trace_error_from_log(ex: Exception) -> str | None:
     """
     # Walk the exception chain looking for a log_file path
     log_file: Path | None = None
-    current: Exception | None = ex
+    current: BaseException | None = ex
     while current is not None:
-        if hasattr(current, "log_file") and current.log_file is not None:
-            log_file = current.log_file
+        current_log_file = getattr(current, "log_file", None)
+        if isinstance(current_log_file, Path):
+            log_file = current_log_file
             break
-        current = getattr(current, "original_exception", None)
+        current = _get_opted_in_original_exception(current)
 
     if log_file is None or not log_file.exists():
         return None
@@ -322,8 +331,8 @@ def _sanitize_error(ex: Exception) -> str:
        `password authentication failed for user "bob"`).  Many Java-based
        connectors emit a generic user-facing `message` while the real
        detail lives only in `internal_message`.
-    2. **Exception chain** - Walks `original_exception` to find the deepest
-       exception with a message more specific than the top-level wrapper.
+    2. **Exception chain** - Walks explicitly opted-in `original_exception`
+       links to find the deepest specific message.
     3. **Top-level exception** - Falls back to `get_message()` or `str()`.
 
     Uses `get_message()` when available (PyAirbyte exceptions) to avoid
@@ -334,10 +343,10 @@ def _sanitize_error(ex: Exception) -> str:
     if trace_msg:
         return f"{type(ex).__name__}: {trace_msg}"
 
-    # 2. Walk original_exception chain to find the deepest specific message.
-    deepest = ex
-    while hasattr(deepest, "original_exception") and deepest.original_exception is not None:
-        deepest = deepest.original_exception
+    # 2. Walk opted-in original_exception links to find the deepest message.
+    deepest: BaseException = ex
+    while (original_exception := _get_opted_in_original_exception(deepest)) is not None:
+        deepest = original_exception
 
     if deepest is not ex:
         deep_msg = (

airbyte/destinations/base.py

@@ -360,7 +360,7 @@ def _write_airbyte_message_stream(
                     connector_name=self.name,
                     log_text=self._last_log_messages,
                     original_exception=ex,
-                ) from None
+                ) from ex
 
 
 __all__ = [

airbyte/exceptions.py

@@ -39,10 +39,10 @@
 from __future__ import annotations
 
 import logging
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from pathlib import Path
 from textwrap import indent
-from typing import TYPE_CHECKING, Any, Protocol
+from typing import TYPE_CHECKING, Any, Protocol, overload
 
 from airbyte._util.meta import is_mcp_mode
 from airbyte.constants import (
@@ -66,6 +66,35 @@
 VERTICAL_SEPARATOR = "\n" + "-" * 60
 
 
+class _CauseDescriptor:
+    @overload
+    def __get__(self, instance: None, owner: type[Any] | None = None) -> None: ...
+
+    @overload
+    def __get__(
+        self,
+        instance: AirbyteLibError,
+        owner: type[Any] | None = None,
+    ) -> BaseException | None: ...
+
+    def __get__(
+        self,
+        instance: AirbyteLibError | None,
+        owner: type[Any] | None = None,
+    ) -> BaseException | None:
+        if instance is None:
+            return None
+        return instance.__cause__
+
+    def __set__(self, instance: AirbyteLibError, value: BaseException | None) -> None:
+        if value is None or value is self:
+            instance.__dict__["_original_exception_opt_in"] = False
+            return
+
+        instance.__cause__ = value
+        instance.__dict__["_original_exception_opt_in"] = True
+
+
 # Base error class
 
 
@@ -80,7 +109,7 @@ class AirbyteLibError(Exception):
     print_full_log: bool = AIRBYTE_PRINT_FULL_ERROR_LOGS
     context: dict[str, Any] | None = None
     message: str | None = None
-    original_exception: Exception | None = None
+    original_exception: _CauseDescriptor = field(default=_CauseDescriptor(), repr=False)
 
     def get_message(self) -> str:
         """Return the best description for the exception.
@@ -136,8 +165,8 @@ def __str__(self) -> str:
 
             exception_str += f"\n    Log output: \n    {indent(self.log_text, '    ')}"
 
-        if self.original_exception:
-            exception_str += VERTICAL_SEPARATOR + f"\nCaused by: {self.original_exception!s}"
+        if getattr(self, "_original_exception_opt_in", False) and self.__cause__ is not None:
+            exception_str += VERTICAL_SEPARATOR + f"\nCaused by: {self.__cause__!s}"
 
         if self.log_file:
             if self.print_full_log:

airbyte/secrets/hydration.py

@@ -122,7 +122,7 @@ def _get_global_secrets_mask() -> list[str]:
         raise AirbyteLibInternalError(
             "Failed to parse spec mask.",
             original_exception=ex,
-        ) from None
+        ) from ex
 
 
 def _get_connector_secrets_mask(

airbyte/sources/util.py

@@ -143,7 +143,7 @@ def get_benchmark_source(
                 message="Invalid number format.",
                 original_exception=ex,
                 input_value=str(num_records),
-            ) from None
+            ) from ex
 
     return get_source(
         name="source-e2e-test",

tests/unit_tests/test_exceptions.py

@@ -2,10 +2,19 @@
 from __future__ import annotations
 
 import inspect
-import airbyte.exceptions as exceptions_module
+from dataclasses import dataclass
+from pathlib import Path
+
 import pytest
 
+import airbyte.exceptions as exceptions_module
+from airbyte._util.destination_smoke_tests import (
+    _extract_trace_error_from_log,
+    _sanitize_error,
+)
 from airbyte.exceptions import (
+    AirbyteCloudError,
+    AirbyteLibError,
     AirbyteMissingWorkspaceContextError,
     AirbyteNoCloudCredentialsError,
 )
@@ -146,5 +155,123 @@ def test_missing_workspace_context_error_guidance(
     assert "Env Vars" not in str(error)
 
 
+def test_original_exception_reflects_raise_from_cause() -> None:
+    inner = ValueError("inner")
+
+    with pytest.raises(AirbyteCloudError) as exc_info:
+        raise AirbyteCloudError(message="m") from inner
+
+    error = exc_info.value
+    assert error.original_exception is inner
+    assert "Caused by:" not in str(error)
+
+
+def test_original_exception_constructor_argument_sets_cause() -> None:
+    inner = ValueError("inner")
+    error = AirbyteCloudError(message="m", original_exception=inner)
+
+    assert error.__cause__ is inner
+    assert error.original_exception is inner
+    assert "Caused by:" in str(error)
+    assert "inner" in str(error)
+
+    replacement = ValueError("replacement")
+    with pytest.raises(AirbyteCloudError) as exc_info:
+        raise error from replacement
+
+    assert exc_info.value.original_exception is replacement
+    assert "Caused by:" in str(exc_info.value)
+    assert "replacement" in str(exc_info.value)
+
+
+def test_original_exception_is_none_without_a_cause() -> None:
+    error = AirbyteCloudError(message="m")
+
+    assert error.original_exception is None
+    assert "Caused by" not in str(error)
+
+
+@pytest.mark.parametrize("error_class", [AirbyteCloudError, AirbyteLibError])
+def test_plain_raise_from_cause_is_not_in_repr(
+    error_class: type[AirbyteLibError],
+) -> None:
+    assert error_class.original_exception is None
+
+    with pytest.raises(error_class) as exc_info:
+        raise error_class(message="m") from ValueError("password=hunter2")
+
+    assert "hunter2" not in repr(exc_info.value)
+
+
+def test_external_dataclass_subclass_with_post_init_accepts_original_exception() -> (
+    None
+):
+    @dataclass
+    class ExternalAirbyteError(AirbyteLibError):
+        post_init_ran: bool = False
+
+        def __post_init__(self) -> None:
+            self.post_init_ran = True
+
+    without_cause = ExternalAirbyteError(message="m")
+    assert without_cause.post_init_ran
+    assert without_cause.original_exception is None
+
+    inner = ValueError("inner")
+    with_cause = ExternalAirbyteError(message="m", original_exception=inner)
+    assert with_cause.post_init_ran
+    assert with_cause.__cause__ is inner
+    assert with_cause.original_exception is inner
+
+
+def test_original_exception_constructor_argument_survives_subclass_post_init() -> None:
+    inner = ValueError("inner")
+    error = AirbyteNoCloudCredentialsError(
+        guidance="guidance",
+        original_exception=inner,
+    )
+
+    assert error.guidance == "guidance"
+    assert error.__cause__ is inner
+    assert error.original_exception is inner
+
+
+def test_sanitize_error_does_not_walk_unopted_raise_from_cause() -> None:
+    inner = ValueError("password=hunter2")
+
+    with pytest.raises(AirbyteCloudError) as exc_info:
+        raise AirbyteCloudError(message="outer") from inner
+
+    error = exc_info.value
+    assert "hunter2" not in _sanitize_error(error)
+    assert "hunter2" not in str(error)
+
+
+def test_sanitize_error_does_not_find_log_file_through_unopted_cause(
+    tmp_path: Path,
+) -> None:
+    class ErrorWithLogFile(Exception):
+        def __init__(self, log_file: Path) -> None:
+            self.log_file = log_file
+
+    log_file = tmp_path / "connector.log"
+    log_file.write_text(
+        '{"type":"TRACE","trace":{"error":{"internal_message":"cause detail"}}}',
+        encoding="utf-8",
+    )
+    inner = ErrorWithLogFile(log_file)
+
+    with pytest.raises(AirbyteCloudError) as exc_info:
+        raise AirbyteCloudError(message="outer") from inner
+
+    assert _extract_trace_error_from_log(exc_info.value) is None
+
+
+def test_sanitize_error_walks_raise_from_chain() -> None:
+    inner = AirbyteCloudError(message="inner")
+    error = AirbyteCloudError(message="outer", original_exception=inner)
+    assert _sanitize_error(error) == "AirbyteCloudError: inner"
+
+
 if __name__ == "__main__":
     pytest.main()