feat(sources): add SourceProxy, identity-preserving schema serialization, and CachedSource graceful degradation

- Add SourceProxy class that preserves content_hash, pipeline_hash, source_id,
  and output_schema for non-reconstructable sources, with bind/unbind delegation
- Add _identity_config() to RootSource base class; all source to_config() methods
  now include identity hashes and serialized schemas
- Implement cross-language Arrow type string parser (parse_arrow_type_string) with
  support for primitives and nested types (list, struct, map, recursive nesting)
- Add proper serialize_schema/deserialize_schema round-trip using Arrow type strings
- Update resolve_source_from_config() with fallback_to_proxy parameter
- CachedSource gracefully serves cached data when inner source is SourceProxy
- Add comprehensive tests for SourceProxy, CachedSource with proxy, schema
  serialization round-trip, and Arrow type string parsing

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: eywalker

src/orcapod/core/sources/__init__.py

@@ -8,6 +8,7 @@
 from .dict_source import DictSource
 from .list_source import ListSource
 from .source_registry import GLOBAL_SOURCE_REGISTRY, SourceRegistry
+from .source_proxy import SourceProxy
 
 __all__ = [
     "RootSource",
@@ -20,5 +21,6 @@
     "DictSource",
     "ListSource",
     "SourceRegistry",
+    "SourceProxy",
     "GLOBAL_SOURCE_REGISTRY",
 ]

src/orcapod/core/sources/arrow_table_source.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from collections.abc import Collection
-from typing import TYPE_CHECKING, Any, Self
+from typing import TYPE_CHECKING, Any
 
 from orcapod.core.sources.base import RootSource
 from orcapod.core.sources.stream_builder import SourceStreamBuilder
@@ -53,10 +53,11 @@ def to_config(self) -> dict[str, Any]:
             "source_type": "arrow_table",
             "tag_columns": list(self._tag_columns),
             "source_id": self.source_id,
+            **self._identity_config(),
         }
 
     @classmethod
-    def from_config(cls, config: dict[str, Any]) -> Self:
+    def from_config(cls, config: dict[str, Any]) -> ArrowTableSource:
         """Not supported — ArrowTableSource cannot be reconstructed from config.
 
         Raises:

src/orcapod/core/sources/base.py

@@ -110,6 +110,24 @@ def resolve_field(self, record_id: str, field_name: str) -> Any:
     # PipelineElementProtocol — schema-only identity (base case of Merkle chain)
     # -------------------------------------------------------------------------
 
+    def _identity_config(self) -> dict[str, Any]:
+        """Return identity fields for inclusion in ``to_config()`` output.
+
+        These fields allow ``SourceProxy`` to be constructed when the source
+        cannot be reconstructed from config, preserving identity hashes and
+        schemas for downstream consumers.
+        """
+        from orcapod.pipeline.serialization import serialize_schema
+
+        tag_schema, packet_schema = self.output_schema()
+        type_converter = self.data_context.type_converter
+        return {
+            "content_hash": self.content_hash().to_string(),
+            "pipeline_hash": self.pipeline_hash().to_string(),
+            "tag_schema": serialize_schema(tag_schema, type_converter),
+            "packet_schema": serialize_schema(packet_schema, type_converter),
+        }
+
     def pipeline_identity_structure(self) -> Any:
         """Return (tag_schema, packet_schema) as the pipeline identity for this
         source.  Schema-only: no data content is included, so sources with

src/orcapod/core/sources/cached_source.py

@@ -54,21 +54,26 @@ def __init__(
         source: SourceProtocol,
         cache_database: ArrowDatabaseProtocol,
         cache_path_prefix: tuple[str, ...] = (),
+        cache_path: tuple[str, ...] | None = None,
+        source_id: str | None = None,
         label: str | None = None,
         data_context: str | contexts.DataContext | None = None,
         config: Config | None = None,
     ) -> None:
         if data_context is None:
             data_context = source.data_context_key
+        if source_id is None:
+            source_id = source.source_id
         super().__init__(
-            source_id=source.source_id,
+            source_id=source_id,
             label=label,
             data_context=data_context,
             config=config,
         )
-        self._source = source
+        self._source: SourceProtocol = source
         self._cache_database = cache_database
         self._cache_path_prefix = cache_path_prefix
+        self._explicit_cache_path = cache_path
         self._cached_stream: ArrowTableStream | None = None
 
     # -------------------------------------------------------------------------
@@ -79,20 +84,28 @@ def to_config(self) -> dict[str, Any]:
         """Serialize this CachedSource configuration to a JSON-compatible dict.
 
         Returns:
-            Dict containing the inner source config, cache database config, and
-            cache path prefix.
+            Dict containing the inner source config, cache database config,
+            cache path prefix, and resolved cache path (for cache-only loading).
         """
         return {
             "source_type": "cached",
             "inner_source": self._source.to_config(),
             "cache_database": self._cache_database.to_config(),
             "cache_path_prefix": list(self._cache_path_prefix),
+            "cache_path": list(self.cache_path),
+            "source_id": self.source_id,
+            **self._identity_config(),
         }
 
     @classmethod
-    def from_config(cls, config: dict[str, Any]) -> "CachedSource":
+    def from_config(cls, config: dict[str, Any]) -> CachedSource:
         """Reconstruct a CachedSource from a config dict.
 
+        If the inner source cannot be resolved (e.g. it requires live data
+        that is unavailable), ``resolve_source_from_config`` returns a
+        ``SourceProxy`` preserving the original source's identity.  The
+        CachedSource can still serve data from its cache database.
+
         Args:
             config: Dict as produced by :meth:`to_config`.
 
@@ -104,12 +117,17 @@ def from_config(cls, config: dict[str, Any]) -> "CachedSource":
             resolve_source_from_config,
         )
 
-        inner_source = resolve_source_from_config(config["inner_source"])
         cache_db = resolve_database_from_config(config["cache_database"])
+        inner_source = resolve_source_from_config(
+            config["inner_source"], fallback_to_proxy=True
+        )
+
         return cls(
             source=inner_source,
             cache_database=cache_db,
             cache_path_prefix=tuple(config.get("cache_path_prefix", ())),
+            cache_path=tuple(config["cache_path"]) if "cache_path" in config else None,
+            source_id=config.get("source_id"),
         )
 
     # -------------------------------------------------------------------------
@@ -126,6 +144,8 @@ def identity_structure(self) -> Any:
     @property
     def cache_path(self) -> tuple[str, ...]:
         """Cache table path, scoped to the source's content hash."""
+        if self._explicit_cache_path is not None:
+            return self._explicit_cache_path
         return self._cache_path_prefix + (
             "source",
             f"node:{self._source.content_hash().to_string()}",
@@ -151,12 +171,12 @@ def keys(
     ) -> tuple[tuple[str, ...], tuple[str, ...]]:
         return self._source.keys(columns=columns, all_info=all_info)
 
-    def _build_merged_stream(self) -> ArrowTableStream:
-        """
-        Run the live source, store new rows in the cache, load all cached
-        rows, and return the merged result as an ArrowTableStream.
+    def _ingest_live_data(self) -> None:
+        """Fetch live data from the source and store new rows in the cache.
+
+        Raises if the source cannot provide data (e.g. an unbound
+        ``SourceProxy``).
         """
-        # Get live source table with source info and system tags
         live_table = self._source.as_table(
             columns={"source": True, "system_tags": True}
         )
@@ -184,16 +204,38 @@ def _build_merged_stream(self) -> ArrowTableStream:
         )
         self._cache_database.flush()
 
-        # Load all cached records (union of current + prior runs)
+    def _build_merged_stream(self) -> ArrowTableStream:
+        """Ingest live data (if available), then return all cached records.
+
+        If the inner source cannot provide data (e.g. an unbound
+        ``SourceProxy``), the method falls back to returning whatever is
+        already stored in the cache database.  If the cache is empty, an
+        empty stream is returned.
+        """
+        try:
+            self._ingest_live_data()
+        except NotImplementedError:
+            logger.info(
+                "Inner source %r cannot provide data; serving from cache only.",
+                self._source.source_id,
+            )
+
         all_records = self._cache_database.get_all_records(self.cache_path)
-        assert all_records is not None, (
-            "Cache should contain records after storing live data."
-        )
+        if all_records is None:
+            all_records = self._empty_table()
 
-        # Build stream from merged table
         tag_keys = self._source.keys()[0]
         return ArrowTableStream(all_records, tag_columns=tag_keys)
 
+    def _empty_table(self) -> pa.Table:
+        """Build an empty Arrow table matching the source's output schema."""
+        tag_schema, packet_schema = self._source.output_schema()
+        merged = dict(tag_schema)
+        merged.update(packet_schema)
+        type_converter = self.data_context.type_converter
+        arrow_schema = type_converter.python_schema_to_arrow_schema(merged)
+        return pa.Table.from_pylist([], schema=arrow_schema)
+
     @property
     def is_stale(self) -> bool:
         """True if the wrapped source has been modified since the last build.

src/orcapod/core/sources/csv_source.py

@@ -64,6 +64,7 @@ def to_config(self) -> dict[str, Any]:
             "system_tag_columns": list(self._system_tag_columns),
             "record_id_column": self._record_id_column,
             "source_id": self.source_id,
+            **self._identity_config(),
         }
 
     @classmethod

src/orcapod/core/sources/data_frame_source.py

@@ -77,6 +77,7 @@ def to_config(self) -> dict[str, Any]:
             "source_type": "data_frame",
             "tag_columns": list(self._tag_columns),
             "source_id": self.source_id,
+            **self._identity_config(),
         }
 
     @classmethod

src/orcapod/core/sources/delta_table_source.py

@@ -75,6 +75,7 @@ def to_config(self) -> dict[str, Any]:
             "system_tag_columns": list(self._system_tag_columns),
             "record_id_column": self._record_id_column,
             "source_id": self.source_id,
+            **self._identity_config(),
         }
 
     @classmethod

src/orcapod/core/sources/dict_source.py

@@ -51,6 +51,7 @@ def to_config(self) -> dict[str, Any]:
             "source_type": "dict",
             "tag_columns": list(self._tag_columns),
             "source_id": self.source_id,
+            **self._identity_config(),
         }
 
     @classmethod

src/orcapod/core/sources/list_source.py

@@ -117,6 +117,7 @@ def to_config(self) -> dict[str, Any]:
             "source_type": "list",
             "name": self.name,
             "source_id": self.source_id,
+            **self._identity_config(),
         }
 
     @classmethod