fix: Add recursive type conversion for nested Mapping objects

- Add _to_builtin_types helper function to recursively convert Mapping-like
  and Sequence-like objects to plain Python types
- This fixes CI test failure where nested objects (like address field) were
  not being properly converted for genson schema inference
- Genson doesn't handle custom Mapping/Sequence implementations properly,
  so we need to convert everything to plain dicts, lists, and primitives

Co-Authored-By: AJ Steers <[email protected]>

Author: devin-ai-integration[bot]

airbyte_cdk/sources/declarative/schema/inferred_schema_loader.py

@@ -3,6 +3,7 @@
 #
 
 from collections.abc import Mapping as ABCMapping
+from collections.abc import Sequence
 from dataclasses import InitVar, dataclass
 from typing import Any, Mapping, Optional
 
@@ -13,6 +14,30 @@
 from airbyte_cdk.utils.schema_inferrer import SchemaInferrer
 
 
+def _to_builtin_types(value: Any) -> Any:
+    """
+    Recursively convert Mapping-like and Sequence-like objects to plain Python types.
+
+    This is necessary because genson's schema inference doesn't handle custom Mapping
+    or Sequence implementations properly. We need to convert everything to plain dicts,
+    lists, and primitive types.
+
+    Args:
+        value: The value to convert
+
+    Returns:
+        The value converted to plain Python types
+    """
+    if isinstance(value, ABCMapping):
+        return {k: _to_builtin_types(v) for k, v in value.items()}
+    elif isinstance(value, (list, tuple)):
+        return [_to_builtin_types(item) for item in value]
+    elif isinstance(value, Sequence) and not isinstance(value, (str, bytes)):
+        return [_to_builtin_types(item) for item in value]
+    else:
+        return value
+
+
 @dataclass
 class InferredSchemaLoader(SchemaLoader):
     """
@@ -63,8 +88,9 @@ def get_json_schema(self) -> Mapping[str, Any]:
                     if record_count >= self.record_sample_size:
                         break
 
-                    if isinstance(record, ABCMapping) and not isinstance(record, dict):
-                        record = dict(record)
+                    # Convert all Mapping-like and Sequence-like objects to plain Python types
+                    # This is necessary because genson doesn't handle custom implementations properly
+                    record = _to_builtin_types(record)
 
                     airbyte_record = AirbyteRecordMessage(
                         stream=self.stream_name,