Add hybrid fs provider

Signed-off-by: Mike Knepper <mknepper@nvidia.com>

Author: mikeknep

packages/data_designer_nemo/src/data_designer_nemo/context.py

@@ -23,7 +23,7 @@
 )
 from data_designer_nemo.errors import NDDError
 from data_designer_nemo.fileset_file_seed_reader import FilesetFileSeedReader
-from data_designer_nemo.fileset_filesystem_provider import FilesetFileSystemProvider
+from data_designer_nemo.fileset_filesystem_provider import FilesetFileSystemProvider, HybridFileSystemProvider
 from data_designer_nemo.model_provider import (
     make_local_first_model_provider_registry,
     make_model_provider_registry,
@@ -84,12 +84,17 @@ async def validate(self, config: dd.DataDesignerConfig) -> list[NDDError]:
         return errors
 
     def get_seed_readers(self) -> list[SeedReader]:
+        # Directory- and FileContents-style seeds may reference either a local
+        # directory or a NeMo Platform fileset in local mode. The engine only
+        # accepts one provider per reader, so we inject a hybrid provider that
+        # resolves each seed path against local disk first, then a fileset.
+        fs_provider = HybridFileSystemProvider(self._sdk, workspace=self._workspace)
         return [
             HuggingFaceSeedReader(),
             LocalFileSeedReader(),
             DataFrameSeedReader(),
-            DirectorySeedReader(),
-            FileContentsSeedReader(),
+            DirectorySeedReader(fs_provider=fs_provider),
+            FileContentsSeedReader(fs_provider=fs_provider),
             AgentRolloutSeedReader(),
             FilesetFileSeedReader(self._sdk),
         ]

packages/data_designer_nemo/src/data_designer_nemo/fileset_filesystem_provider.py

@@ -1,9 +1,11 @@
 # SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
 
-from pathlib import PurePosixPath
+from pathlib import Path, PurePosixPath
 
 from data_designer.engine.resources.seed_reader import (
+    FileSystemProvider,
+    LocalFileSystemProvider,
     SeedReaderConfigError,
     SeedReaderError,
     SeedReaderFileSystemContext,
@@ -89,3 +91,42 @@ def _parse(self, runtime_path: str) -> tuple[str, str, str]:
             return parse_fileset_ref(runtime_path, workspace_fallback=self._workspace)
         except FilesetPathError as error:
             raise SeedReaderError(f"🛑 Invalid fileset seed source path {runtime_path!r}: {error}") from error
+
+
+class HybridFileSystemProvider:
+    """Filesystem provider that resolves a seed path against local disk first, then a fileset.
+
+    In local mode a directory-style seed source may point at either a directory on
+    the local filesystem or a NeMo Platform fileset, and the engine only lets us
+    inject a single provider per seed reader. We route per path: if the path
+    resolves to an existing local directory we serve it from disk, otherwise we
+    treat it as a fileset reference. This mirrors the local-first model-provider
+    resolution strategy (locally-defined providers first, Inference Gateway as the
+    fallback).
+    """
+
+    def __init__(
+        self,
+        sdk: NeMoPlatform | AsyncNeMoPlatform,
+        *,
+        workspace: str,
+        validated_roots: set[str] | None = None,
+    ) -> None:
+        self._local = LocalFileSystemProvider()
+        self._fileset = FilesetFileSystemProvider(sdk, workspace=workspace, validated_roots=validated_roots)
+
+    def create_context(self, *, runtime_path: str) -> SeedReaderFileSystemContext:
+        return self._route(runtime_path).create_context(runtime_path=runtime_path)
+
+    def ensure_root_exists(self, *, runtime_path: str) -> None:
+        self._route(runtime_path).ensure_root_exists(runtime_path=runtime_path)
+
+    def _route(self, runtime_path: str) -> FileSystemProvider:
+        return self._local if _is_local_directory(runtime_path) else self._fileset
+
+
+def _is_local_directory(runtime_path: str) -> bool:
+    try:
+        return Path(runtime_path).expanduser().is_dir()
+    except (OSError, ValueError, RuntimeError):
+        return False

packages/data_designer_nemo/tests/unit/test_fileset_filesystem_provider.py

@@ -1,11 +1,17 @@
 # SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
 
+from pathlib import Path
 from unittest.mock import Mock, patch
 
 import pytest
-from data_designer.engine.resources.seed_reader import SeedReaderConfigError
-from data_designer_nemo.fileset_filesystem_provider import FilesetFileSystemProvider
+from data_designer.engine.resources.seed_reader import (
+    DirectorySeedReader,
+    FileContentsSeedReader,
+    SeedReaderConfigError,
+)
+from data_designer_nemo.context import LocalDataDesignerContext
+from data_designer_nemo.fileset_filesystem_provider import FilesetFileSystemProvider, HybridFileSystemProvider
 
 
 def test_create_context_roots_reader_in_canonical_fileset_ref() -> None:
@@ -44,3 +50,49 @@ def test_ensure_root_exists_reports_missing_fileset_path() -> None:
             provider.ensure_root_exists(runtime_path="docs#corpus")
 
     assert fs_class.return_value.exists.call_count == 2
+
+
+def test_hybrid_routes_existing_local_directory_to_disk(tmp_path: Path) -> None:
+    sdk = Mock()
+    provider = HybridFileSystemProvider(sdk, workspace="default")
+
+    with patch("data_designer_nemo.fileset_filesystem_provider.FilesetFileSystem") as fs_class:
+        context = provider.create_context(runtime_path=str(tmp_path))
+        provider.ensure_root_exists(runtime_path=str(tmp_path))
+
+    assert context.root_path == tmp_path.resolve()
+    fs_class.assert_not_called()
+
+
+def test_hybrid_routes_non_local_path_to_fileset() -> None:
+    sdk = Mock()
+    provider = HybridFileSystemProvider(sdk, workspace="default")
+
+    with patch("data_designer_nemo.fileset_filesystem_provider.FilesetFileSystem") as fs_class:
+        fs_class.return_value.async_impl = True
+        fs_class.return_value.asynchronous = False
+        context = provider.create_context(runtime_path="docs#corpus")
+
+    fs_class.assert_called_once_with(sdk)
+    assert str(context.root_path) == "default/docs#corpus"
+
+
+def test_hybrid_ensure_root_exists_validates_fileset_for_non_local_path() -> None:
+    sdk = Mock()
+    provider = HybridFileSystemProvider(sdk, workspace="default")
+
+    with patch("data_designer_nemo.fileset_filesystem_provider.FilesetFileSystem") as fs_class:
+        fs_class.return_value.exists.side_effect = [False, True]
+
+        with pytest.raises(SeedReaderConfigError, match="Path 'corpus' not found in fileset 'default/docs'"):
+            provider.ensure_root_exists(runtime_path="docs#corpus")
+
+    assert fs_class.return_value.exists.call_count == 2
+
+
+def test_local_context_wires_hybrid_provider_into_filesystem_readers() -> None:
+    readers = LocalDataDesignerContext(Mock(), "default").get_seed_readers()
+
+    fs_readers = [r for r in readers if isinstance(r, DirectorySeedReader | FileContentsSeedReader)]
+    assert len(fs_readers) == 2
+    assert all(isinstance(r._fs_provider, HybridFileSystemProvider) for r in fs_readers)