fix(ingest): preserve cloud workspace URIs in CollectionBuilder base class

CollectionBuilder.__init__ unconditionally wrapped workspace_dir in
Path(), corrupting cloud URIs like s3://bucket/path into s3:/bucket/path.

Keep cloud URIs (containing ://) as strings; continue using Path for
local filesystem paths. This fixes the root cause at the base class
level, complementing the call-site guard added in 1977f91.

Author: print-sid8

src/rasteret/ingest/base.py

@@ -44,7 +44,16 @@ def __init__(
     ) -> None:
         self.name = name
         self.data_source = data_source
-        self.workspace_dir = Path(workspace_dir) if workspace_dir else None
+        if workspace_dir is None:
+            self.workspace_dir: str | Path | None = None
+        elif isinstance(workspace_dir, Path):
+            self.workspace_dir = workspace_dir
+        else:
+            ws = str(workspace_dir)
+            if "://" in ws and not ws.startswith("file://"):
+                self.workspace_dir = ws
+            else:
+                self.workspace_dir = Path(ws)
 
     @abstractmethod
     def build(self, **kwargs: Any) -> "Collection":

src/rasteret/tests/test_ingest.py

@@ -130,6 +130,27 @@ def test_load_valid_manifest(self, tmp_path):
         assert isinstance(collection, Collection)
         assert collection.name == "manifest"
 
+    def test_workspace_dir_cloud_uri_not_mangled(self, tmp_path, monkeypatch):
+        path = tmp_path / "manifest.parquet"
+        _write_manifest(path)
+        captured: dict[str, object] = {}
+
+        def _fake_build_collection_from_table(table: pa.Table, **kwargs):
+            captured["workspace_dir"] = kwargs.get("workspace_dir")
+            return Collection(dataset=ds.dataset(table), name="manifest")
+
+        monkeypatch.setattr(
+            "rasteret.ingest.parquet_record_table.build_collection_from_table",
+            _fake_build_collection_from_table,
+        )
+
+        builder = RecordTableBuilder(
+            path,
+            workspace_dir="s3://demo-bucket/workspace_records",
+        )
+        builder.build()
+        assert captured["workspace_dir"] == "s3://demo-bucket/workspace_records"
+
     def test_column_remapping(self, tmp_path):
         table = pa.table(
             {