chore: merge main into scopes-array-schema to resolve conflicts

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:

airbyte_cdk/sources/declarative/interpolation/jinja.py

@@ -124,8 +124,8 @@ def eval(
 
     def _literal_eval(self, result: Optional[str], valid_types: Optional[Tuple[Type[Any]]]) -> Any:
         try:
-            evaluated = ast.literal_eval(result)  # type: ignore # literal_eval is able to handle None
-        except (ValueError, SyntaxError):
+            evaluated = ast.literal_eval(result)  # type: ignore # result may be None; on error we return it unchanged
+        except (ValueError, SyntaxError, TypeError):
             return result
         if (not valid_types and not isinstance(evaluated, complex)) or (
             valid_types and isinstance(evaluated, valid_types)

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/declarative/interpolation/test_jinja.py

@@ -345,6 +345,16 @@ def test_interpolation_private_partition_attribute():
     assert actual_output == expected_output
 
 
+def test_literal_eval_handles_unhashable_set_typeerror():
+    """Test that _literal_eval gracefully handles TypeError from ast.literal_eval for nested sets."""
+    # ast.literal_eval("{{'web'}, {'discover'}}") raises TypeError: unhashable type: 'set'
+    # The interpolation should return the string as-is instead of propagating the error.
+    config = {"query": "{{'web'}, {'discover'}}"}
+    s = "{{ config['query'] }}"
+    val = interpolation.eval(s, config)
+    assert val == "{{'web'}, {'discover'}}"
+
+
 def test_given_complex_when_eval_then_return_string():
     s = "9173710294242221J"
     config = {}

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]