diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
index 9d0534c805..066fe33aac 100644
--- a/.github/workflows/ci.yaml
+++ b/.github/workflows/ci.yaml
@@ -337,12 +337,16 @@ jobs:
         run: make test-e2e
         env:
           _TYPER_FORCE_DISABLE_TERMINAL: "1"
-          E2E_SERVICES_LOG: ${{ runner.temp }}/services.log
+          E2E_SERVICES_LOG_DIR: ${{ runner.temp }}/e2e-services-logs
+          NGC_API_KEY: ${{ secrets.NGC_REGISTRY_READ_TOKEN }}
+          HF_TOKEN: ${{ secrets.HF_TOKEN }}
       - name: Dump server logs
         if: always()
         run: |
-          echo "::group::Server log"
-          cat "${{ runner.temp }}/services.log" 2>/dev/null || echo "No server log found"
+          echo "::group::Server logs"
+          for f in "${{ runner.temp }}/e2e-services-logs"/*.log; do
+            [ -f "$f" ] && echo "--- $(basename "$f") ---" && cat "$f" || echo "No server logs found"
+          done
           echo "::endgroup::"
       - name: Upload test artifacts
         if: always()
@@ -352,7 +356,7 @@ jobs:
           retention-days: 30
           path: |
             report.xml
-            ${{ runner.temp }}/services.log
+            ${{ runner.temp }}/e2e-services-logs/
 
   # Required-check pin: branch protection should reference this aggregator
   # rather than the per-row matrix jobs, so the matrix can grow or shrink
diff --git a/e2e/conftest.py b/e2e/conftest.py
index f9de40bfc2..a42a9741fd 100644
--- a/e2e/conftest.py
+++ b/e2e/conftest.py
@@ -23,7 +23,6 @@
 import socket
 import subprocess
 import sys
-import tempfile
 import time
 import uuid
 from collections.abc import Iterator
@@ -54,7 +53,36 @@ def pytest_configure(config: pytest.Config) -> None:
 
 _HEALTH_TIMEOUT = 60
 _HEALTH_POLL_INTERVAL = 1.0
-_SERVICES_LOG = Path(os.environ.get("E2E_SERVICES_LOG", os.path.join(tempfile.gettempdir(), "services.log")))
+
+# Number of log lines to dump from the services log on test failure.
+_TAIL_LINES_ON_FAILURE = 100
+
+_services_log_key = pytest.StashKey[Path]()
+
+
+@pytest.fixture(scope="session")
+def services_log_path(request: pytest.FixtureRequest, tmp_path_factory: pytest.TempPathFactory) -> Path:
+    """Return a unique services log path for this session.
+
+    ``E2E_SERVICES_LOG_DIR`` (if set) is treated as a **directory**; in CI
+    the job uploads everything under it as artifacts.  When unset we
+    fall back to a pytest-managed temp directory.  Either way, each
+    session writes to a UUID-named file inside the directory so
+    parallel workers never clobber each other.
+
+    The path is stashed on the session so the
+    ``pytest_runtest_makereport`` hook can read it without requesting
+    the fixture.
+    """
+    log_dir = os.environ.get("E2E_SERVICES_LOG_DIR")
+    if log_dir:
+        directory = Path(log_dir)
+        directory.mkdir(parents=True, exist_ok=True)
+    else:
+        directory = tmp_path_factory.mktemp("e2e-services-logs")
+    path = directory / f"services-{uuid.uuid4().hex[:8]}.log"
+    request.session.stash[_services_log_key] = path
+    return path
 
 
 def _find_free_port() -> int:
@@ -99,8 +127,37 @@ def background_process(args: list[str], stdout: IO[Any] | None = None) -> Iterat
             proc.wait(timeout=5)
 
 
+# ---- Services log tail on failure ------------------------------------------
+
+
+@pytest.hookimpl(hookwrapper=True)
+def pytest_runtest_makereport(item: pytest.Item, call: pytest.CallInfo):  # noqa: ARG001
+    """Append the services log tail to the report when a test fails.
+
+    This hook is the pytest-sanctioned way to add extra sections to test
+    reports (``report.sections``).  Fixtures cannot do this because they
+    don't have access to the report object.
+    """
+    outcome = yield
+    report = outcome.get_result()
+
+    if not report.failed:
+        return
+
+    log_path = item.session.stash.get(_services_log_key, None)
+    if log_path and log_path.exists():
+        lines = log_path.read_text().splitlines(keepends=True)
+        tail = lines[-_TAIL_LINES_ON_FAILURE:]
+        if tail:
+            header = f"--- services log (last {len(tail)} lines) [{log_path}] ---"
+            report.sections.append(("Services Log", f"{header}\n{''.join(tail)}"))
+
+
+# ---- Fixtures --------------------------------------------------------------
+
+
 @pytest.fixture(scope="session")
-def _services() -> Iterator[str]:
+def _services(services_log_path: Path) -> Iterator[str]:
     """Spawn ``nemo services run`` and yield the base URL.
 
     Skipped when ``NMP_BASE_URL`` is already set (external services).
@@ -124,7 +181,7 @@ def _services() -> Iterator[str]:
 
     logger.info("Starting nemo services on port %d", port)
 
-    log_path = _SERVICES_LOG
+    log_path = services_log_path
     with open(log_path, "w") as log_file, background_process(args, stdout=log_file) as proc:
         if not _wait_for_healthy(url):
             pytest.fail(
diff --git a/e2e/test_files.py b/e2e/files/test_files.py
similarity index 100%
rename from e2e/test_files.py
rename to e2e/files/test_files.py
diff --git a/e2e/files/test_storage_backends.py b/e2e/files/test_storage_backends.py
new file mode 100644
index 0000000000..39f9125a0e
--- /dev/null
+++ b/e2e/files/test_storage_backends.py
@@ -0,0 +1,312 @@
+"""E2E tests for external storage backends (NGC, Hugging Face).
+
+These tests verify that the files service can create filesets backed by
+external storage providers and read files from them via the SDK.
+
+NGC tests require ``NGC_API_KEY`` in the environment and are skipped
+otherwise.  Hugging Face tests use a small public repo; when ``HF_TOKEN``
+is set the request is authenticated (avoids rate-limits in CI).
+"""
+
+import os
+import tempfile
+import uuid
+from collections.abc import Iterator
+from pathlib import Path
+
+import pytest
+from nemo_platform import BadRequestError, NeMoPlatform
+from nemo_platform.types.files import HuggingfaceStorageConfigParam, NGCStorageConfigParam
+
+# ---------------------------------------------------------------------------
+# NGC configuration
+# ---------------------------------------------------------------------------
+NGC_API_KEY_ENV = "NGC_API_KEY"
+
+NGC_ORG = "nvidia"
+NGC_TEAM = "nemo-microservices"
+NGC_TARGET = "nemo-microservices-quickstart"
+NGC_TARGET_TYPE = "resource"
+
+# ---------------------------------------------------------------------------
+# Hugging Face configuration — small public model
+# ---------------------------------------------------------------------------
+HF_TOKEN_ENV = "HF_TOKEN"
+
+HF_REPO_ID = "hf-internal-testing/tiny-random-bert"
+HF_REPO_TYPE = "model"
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+def ngc_api_key() -> str:
+    """Return the NGC API key from the environment."""
+    key = os.environ.get(NGC_API_KEY_ENV)
+    assert key, f"{NGC_API_KEY_ENV} must be set"
+    return key
+
+
+@pytest.fixture
+def ngc_secret(sdk: NeMoPlatform, workspace: str, ngc_api_key: str) -> Iterator[str]:
+    """Create a secret containing the NGC API key, cleaned up after test."""
+    secret_name = f"e2e-ngc-key-{uuid.uuid4().hex[:8]}"
+    sdk.secrets.create(workspace=workspace, name=secret_name, value=ngc_api_key)
+    yield secret_name
+    try:
+        sdk.secrets.delete(workspace=workspace, name=secret_name)
+    except Exception:
+        pass  # Best-effort cleanup; the workspace is deleted anyway
+
+
+@pytest.fixture
+def ngc_fileset(sdk: NeMoPlatform, workspace: str, ngc_secret: str) -> Iterator[str]:
+    """Create an NGC-backed fileset, cleaned up after test."""
+    fileset_name = f"e2e-ngc-fs-{uuid.uuid4().hex[:8]}"
+    sdk.files.filesets.create(
+        workspace=workspace,
+        name=fileset_name,
+        description="E2E test NGC-backed fileset",
+        storage=NGCStorageConfigParam(
+            api_key_secret=ngc_secret,
+            org=NGC_ORG,
+            team=NGC_TEAM,
+            target=NGC_TARGET,
+            target_type=NGC_TARGET_TYPE,
+        ),
+    )
+    yield fileset_name
+    try:
+        sdk.files.filesets.delete(fileset_name, workspace=workspace)
+    except Exception:
+        pass  # Best-effort cleanup; the workspace is deleted anyway
+
+
+@pytest.fixture
+def hf_token() -> str:
+    """Return the HF token from the environment."""
+    token = os.environ.get(HF_TOKEN_ENV)
+    assert token, f"{HF_TOKEN_ENV} must be set"
+    return token
+
+
+@pytest.fixture
+def hf_secret(sdk: NeMoPlatform, workspace: str, hf_token: str) -> Iterator[str]:
+    """Create a secret containing the HF token, cleaned up after test."""
+    secret_name = f"e2e-hf-tok-{uuid.uuid4().hex[:8]}"
+    sdk.secrets.create(workspace=workspace, name=secret_name, value=hf_token)
+    yield secret_name
+    try:
+        sdk.secrets.delete(workspace=workspace, name=secret_name)
+    except Exception:
+        pass  # Best-effort cleanup; the workspace is deleted anyway
+
+
+@pytest.fixture
+def hf_fileset(sdk: NeMoPlatform, workspace: str, hf_secret: str) -> Iterator[str]:
+    """Create a Hugging Face-backed fileset, cleaned up after test."""
+    fileset_name = f"e2e-hf-fs-{uuid.uuid4().hex[:8]}"
+
+    storage = HuggingfaceStorageConfigParam(
+        repo_id=HF_REPO_ID,
+        repo_type=HF_REPO_TYPE,
+        token_secret=hf_secret,
+    )
+
+    sdk.files.filesets.create(
+        workspace=workspace,
+        name=fileset_name,
+        description="E2E test HF-backed fileset",
+        storage=storage,
+    )
+    yield fileset_name
+    try:
+        sdk.files.filesets.delete(fileset_name, workspace=workspace)
+    except Exception:
+        pass  # Best-effort cleanup; the workspace is deleted anyway
+
+
+# ===================================================================
+# NGC tests
+# ===================================================================
+
+
+@pytest.mark.skipif(not os.environ.get(NGC_API_KEY_ENV), reason=f"{NGC_API_KEY_ENV} not set")
+class TestNGCFileset:
+    """Tests for NGC-backed filesets."""
+
+    def test_list_files(self, sdk: NeMoPlatform, workspace: str, ngc_fileset: str):
+        """Listing an NGC-backed fileset returns files with paths and sizes."""
+        files = sdk.files.list(fileset=ngc_fileset, workspace=workspace)
+        assert len(files.data) > 0, "NGC fileset should contain at least one file"
+
+        for f in files.data:
+            assert f.path, "Each file should have a path"
+            assert f.size > 0, "Each file should have a non-zero size"
+
+    def test_download_file(self, sdk: NeMoPlatform, workspace: str, ngc_fileset: str):
+        """Downloading the smallest file from an NGC fileset succeeds and size matches."""
+        files = sdk.files.list(fileset=ngc_fileset, workspace=workspace)
+        assert len(files.data) > 0
+
+        target = min(files.data, key=lambda f: f.size)
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            local_path = Path(tmpdir) / target.path.replace("/", "_")
+            sdk.files.download(
+                fileset=ngc_fileset,
+                workspace=workspace,
+                remote_path=target.path,
+                local_path=str(local_path),
+            )
+            assert local_path.exists()
+            assert local_path.stat().st_size == target.size
+
+    def test_cache_status(self, sdk: NeMoPlatform, workspace: str, ngc_fileset: str):
+        """NGC-backed files report a cacheable status."""
+        files = sdk.files.list(
+            fileset=ngc_fileset,
+            workspace=workspace,
+            include_cache_status=True,
+        )
+        assert len(files.data) > 0
+
+        for f in files.data:
+            assert f.cache_status is not None
+            assert f.cache_status != "not_cacheable"
+
+    # -- error cases --
+
+    @pytest.mark.parametrize(
+        ("secret_value", "storage_overrides", "match"),
+        [
+            pytest.param(
+                "not-a-real-key",
+                {},
+                "Invalid API key. Legacy NGC keys are not supported.",
+                id="invalid-key-prefix",
+            ),
+            pytest.param(
+                None,
+                {"org": "nvidian", "team": "nemo-llm", "target": "nemo-platform-quickstart"},
+                "Error creating NGC storage backend:",
+                id="wrong-org",
+            ),
+            pytest.param(
+                None,
+                {"target": "this-resource-does-not-exist-12345"},
+                "Failed to access NGC resource this-resource-does-not-exist-12345",
+                id="nonexistent-resource",
+            ),
+        ],
+    )
+    def test_create_error(
+        self,
+        sdk: NeMoPlatform,
+        workspace: str,
+        ngc_api_key: str,
+        secret_value: str | None,
+        storage_overrides: dict,
+        match: str,
+    ):
+        """Bad NGC configurations are rejected with 400."""
+        value = secret_value if secret_value is not None else ngc_api_key
+        secret_name = f"e2e-ngc-err-{uuid.uuid4().hex[:8]}"
+        sdk.secrets.create(workspace=workspace, name=secret_name, value=value)
+        try:
+            storage = NGCStorageConfigParam(
+                api_key_secret=secret_name,
+                org=storage_overrides.get("org", NGC_ORG),
+                team=storage_overrides.get("team", NGC_TEAM),
+                target=storage_overrides.get("target", NGC_TARGET),
+                target_type=NGC_TARGET_TYPE,
+            )
+            with pytest.raises(BadRequestError, match=match):
+                sdk.files.filesets.create(
+                    workspace=workspace,
+                    name=f"e2e-ngc-err-{uuid.uuid4().hex[:8]}",
+                    storage=storage,
+                )
+        finally:
+            sdk.secrets.delete(workspace=workspace, name=secret_name)
+
+    def test_create_error_nonexistent_secret(self, sdk: NeMoPlatform, workspace: str):
+        """Referencing a secret that doesn't exist is rejected with 400."""
+        with pytest.raises(BadRequestError, match="Secret not found:"):
+            sdk.files.filesets.create(
+                workspace=workspace,
+                name=f"e2e-ngc-err-{uuid.uuid4().hex[:8]}",
+                storage=NGCStorageConfigParam(
+                    api_key_secret="no-such-secret-99999",
+                    org=NGC_ORG,
+                    team=NGC_TEAM,
+                    target=NGC_TARGET,
+                    target_type=NGC_TARGET_TYPE,
+                ),
+            )
+
+
+# ===================================================================
+# Hugging Face tests
+# ===================================================================
+
+
+@pytest.mark.skipif(not os.environ.get(HF_TOKEN_ENV), reason=f"{HF_TOKEN_ENV} not set")
+class TestHuggingFaceFileset:
+    """Tests for Hugging Face-backed filesets."""
+
+    def test_list_files(self, sdk: NeMoPlatform, workspace: str, hf_fileset: str):
+        """Listing an HF-backed fileset returns files with paths and sizes."""
+        files = sdk.files.list(fileset=hf_fileset, workspace=workspace)
+        assert len(files.data) > 0, "HF fileset should contain at least one file"
+
+        for f in files.data:
+            assert f.path, "Each file should have a path"
+            assert f.size > 0, "Each file should have a non-zero size"
+
+    def test_download_file(self, sdk: NeMoPlatform, workspace: str, hf_fileset: str):
+        """Downloading the smallest file from an HF fileset succeeds and size matches."""
+        files = sdk.files.list(fileset=hf_fileset, workspace=workspace)
+        assert len(files.data) > 0
+
+        target = min(files.data, key=lambda f: f.size)
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            local_path = Path(tmpdir) / target.path.replace("/", "_")
+            sdk.files.download(
+                fileset=hf_fileset,
+                workspace=workspace,
+                remote_path=target.path,
+                local_path=str(local_path),
+            )
+            assert local_path.exists()
+            assert local_path.stat().st_size == target.size
+
+    def test_cache_status(self, sdk: NeMoPlatform, workspace: str, hf_fileset: str):
+        """HF-backed files report a cacheable status."""
+        files = sdk.files.list(
+            fileset=hf_fileset,
+            workspace=workspace,
+            include_cache_status=True,
+        )
+        assert len(files.data) > 0
+
+        for f in files.data:
+            assert f.cache_status is not None
+            assert f.cache_status != "not_cacheable"
+
+    # -- error cases --
+
+    def test_error_nonexistent_repo(self, sdk: NeMoPlatform, workspace: str):
+        """Pointing at a repo that doesn't exist is rejected with 400."""
+        with pytest.raises(BadRequestError):
+            sdk.files.filesets.create(
+                workspace=workspace,
+                name=f"e2e-hf-err-{uuid.uuid4().hex[:8]}",
+                storage=HuggingfaceStorageConfigParam(
+                    repo_id="this-org-does-not-exist/this-repo-does-not-exist-12345",
+                    repo_type="model",
+                ),
+            )
diff --git a/services/core/files/src/nmp/core/files/app/backends/ngc.py b/services/core/files/src/nmp/core/files/app/backends/ngc.py
index 06e6bcad26..a081f6affd 100644
--- a/services/core/files/src/nmp/core/files/app/backends/ngc.py
+++ b/services/core/files/src/nmp/core/files/app/backends/ngc.py
@@ -15,7 +15,7 @@
 import aiohttp
 from anyio import to_thread
 from ngcbase.constants import SCOPED_KEY_PREFIX
-from ngcbase.errors import ResourceNotFoundException
+from ngcbase.errors import NgcException, ResourceNotFoundException
 from ngcsdk import Client
 from nmp.common.files.storage_config import NGCStorageConfig as NGCStorageConfig
 from nmp.core.files.app.backends.base import (
@@ -91,7 +91,7 @@ def _configure() -> Client:
 
         try:
             self._client = await to_thread.run_sync(_configure)
-        except ValueError as exc:
+        except (ValueError, NgcException) as exc:
             raise NGCBackendError(f"Error creating NGC storage backend: [{str(exc)}]") from exc
 
         return self._client
@@ -264,17 +264,17 @@ async def validate_storage(self):
         """Validate that we can access the NGC asset."""
         validate_external_host(self.config.host)
 
-        registry_api = await self._get_registry_api()
-        target = await self._get_target()
-        target_with_version = await self._get_target_with_version()
-
         try:
+            registry_api = await self._get_registry_api()
+            target = await self._get_target()
             await to_thread.run_sync(registry_api.info, target)
+        except NGCBackendError:
+            raise
         except ResourceNotFoundException as exc:
-            raise NGCBackendError(f"NGC {self.config.target_type} not found: {target_with_version}") from exc
+            raise NGCBackendError(f"NGC {self.config.target_type} not found: {self.config.target}") from exc
         except Exception as exc:
             raise NGCBackendError(
-                f"Failed to access NGC {self.config.target_type} {target_with_version} [{str(exc)}]"
+                f"Failed to access NGC {self.config.target_type} {self.config.target} [{exc}]"
             ) from exc
 
     async def upload(