fix(cdk): Preserve error message in concurrent source exception handling (AI-Triage PR) (#937)

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: bot_apk <apk@cognition.ai>

Author: devin-ai-integration[bot]

airbyte_cdk/sources/concurrent_source/concurrent_read_processor.py

@@ -173,7 +173,9 @@ def on_exception(self, exception: StreamThreadException) -> Iterable[AirbyteMess
             yield exception.exception.as_airbyte_message(stream_descriptor=stream_descriptor)
         else:
             yield AirbyteTracedException.from_exception(
-                exception, stream_descriptor=stream_descriptor
+                exception.exception,
+                stream_descriptor=stream_descriptor,
+                message=f"An unexpected error occurred in stream {exception.stream_name}: {type(exception.exception).__name__}",
             ).as_airbyte_message()
 
     def _flag_exception(self, stream_name: str, exception: Exception) -> None:

unit_tests/connector_builder/test_connector_builder_handler.py

@@ -1288,13 +1288,13 @@ def test_handle_read_external_requests(deployment_mode, url_base, expected_error
         pytest.param(
             "CLOUD",
             "https://10.0.27.27/tokens/bearer",
-            "StreamThreadException",
+            "Error while refreshing access token",
             id="test_cloud_read_with_private_endpoint",
         ),
         pytest.param(
             "CLOUD",
             "http://unsecured.protocol/tokens/bearer",
-            "StreamThreadException",
+            "Invalid Protocol Scheme",
             id="test_cloud_read_with_unsecured_endpoint",
         ),
         pytest.param(

unit_tests/sources/streams/concurrent/test_concurrent_read_processor.py

@@ -16,6 +16,7 @@
     AirbyteStreamStatus,
     AirbyteStreamStatusTraceMessage,
     AirbyteTraceMessage,
+    FailureType,
     StreamDescriptor,
     SyncMode,
     TraceType,
@@ -576,7 +577,11 @@ def test_on_exception_return_trace_message_and_on_stream_complete_return_stream_
 
         exception_messages = list(handler.on_exception(exception))
         assert len(exception_messages) == 1
-        assert "StreamThreadException" in exception_messages[0].trace.error.stack_trace
+        assert "RuntimeError" in exception_messages[0].trace.error.stack_trace
+        assert (
+            exception_messages[0].trace.error.message
+            == f"An unexpected error occurred in stream {_STREAM_NAME}: RuntimeError"
+        )
 
         assert list(
             handler.on_partition_complete_sentinel(
@@ -761,6 +766,52 @@ def test_is_done_is_true_if_all_partitions_are_closed_and_no_streams_are_generat
 
         assert handler.is_done()
 
+    @freezegun.freeze_time("2020-01-01T00:00:00")
+    def test_on_exception_non_ate_uses_templated_message_with_correct_failure_type(self):
+        """Regression test: non-ATE exceptions on Path B produce a safe templated message, not the generic fallback."""
+        stream_instances_to_read_from = [self._stream, self._another_stream]
+
+        handler = ConcurrentReadProcessor(
+            stream_instances_to_read_from,
+            self._partition_enqueuer,
+            self._thread_pool_manager,
+            self._logger,
+            self._slice_logger,
+            self._message_repository,
+            self._partition_reader,
+        )
+
+        handler.start_next_partition_generator()
+        handler.on_partition(self._an_open_partition)
+        list(
+            handler.on_partition_generation_completed(
+                PartitionGenerationCompletedSentinel(self._stream)
+            )
+        )
+        list(
+            handler.on_partition_generation_completed(
+                PartitionGenerationCompletedSentinel(self._another_stream)
+            )
+        )
+
+        inner_exception = ValueError("some internal detail: SELECT * FROM secrets")
+        exception = StreamThreadException(inner_exception, _STREAM_NAME)
+
+        exception_messages = list(handler.on_exception(exception))
+        assert len(exception_messages) == 1
+
+        trace_error = exception_messages[0].trace.error
+        # User-facing message uses safe template (no raw exception text)
+        assert (
+            trace_error.message
+            == f"An unexpected error occurred in stream {_STREAM_NAME}: ValueError"
+        )
+        # Stack trace comes from the inner exception, not the StreamThreadException wrapper
+        assert "ValueError" in trace_error.stack_trace
+        assert "StreamThreadException" not in trace_error.stack_trace
+        # failure_type defaults to system_error for unhandled exceptions
+        assert trace_error.failure_type == FailureType.system_error
+
     @freezegun.freeze_time("2020-01-01T00:00:00")
     def test_start_next_partition_generator(self):
         stream_instances_to_read_from = [self._stream]