fix(agents): stage agent spec fileset in subprocess deployments

plugins/nemo-agents/src/nemo_agents_plugin/runner/fabric_artifact_staging.py

@@ -1,12 +1,14 @@
 # SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
 
-"""Stage Fabric agent-spec fileset artifacts for container deployments."""
+"""Stage Fabric agent-spec fileset artifacts for container and subprocess deployments."""
 
 from __future__ import annotations
 
 import logging
+import shutil
 import tempfile
+from collections.abc import Collection
 from pathlib import Path, PurePosixPath
 from typing import Any, Protocol
 
@@ -38,6 +40,89 @@ async def download(
     ) -> None: ...
 
 
+async def stage_fabric_spec_dir(
+    *,
+    workspace: str,
+    agent_name: str,
+    agent_config: dict[str, Any],
+    base_dir: Path,
+    sdk: _FilesDownloader | None,
+) -> None:
+    """Download the agent spec fileset into *base_dir* for subprocess deployments.
+
+    Mirrors :func:`stage_fabric_spec_config_files` for a runtime that reads the
+    agent root off local disk rather than through a ConfigMap. An unavailable
+    fileset is not fatal — config-only agents deploy from ``agent.yaml`` alone —
+    but a config referencing skills still needs them staged.
+
+    The container byte cap does not apply here: it bounds ConfigMap and env
+    delivery, neither of which is in this path.
+
+    *base_dir* is reused when the controller restarts a subprocess deployment,
+    so previously staged files are cleared first. Otherwise a file dropped from
+    the fileset would survive locally, and stale skills could satisfy validation
+    for a deployment that staged nothing.
+    """
+    _clear_staged_tree(base_dir, _runtime_dir_names(agent_config))
+
+    if not agent_name or sdk is None:
+        validate_referenced_skill_paths(agent_config, _staged_relative_paths(base_dir))
+        return
+
+    fileset_name = agent_spec_fileset_name(agent_name)
+    try:
+        await sdk.download(local_path=str(base_dir), fileset=fileset_name, workspace=workspace)
+    except (FileNotFoundError, PlatformNotFoundError, PluginClientNotFoundError) as exc:
+        logger.info(
+            "Agent spec fileset %s/%s unavailable (%s); using inline agent.yaml only",
+            workspace,
+            fileset_name,
+            exc,
+        )
+    validate_referenced_skill_paths(agent_config, _staged_relative_paths(base_dir))
+
+
+def _runtime_dir_names(agent_config: dict[str, Any]) -> set[str]:
+    """Return top-level names under the agent root that the runtime owns, not staging.
+
+    ``environment.workspace`` and ``environment.artifacts`` hold a live agent's
+    output (Relay telemetry among it) and resolve relative to ``agent.yaml``, so
+    restaging must not delete them.
+    """
+    environment = agent_config.get("environment")
+    if not isinstance(environment, dict):
+        return set()
+
+    names: set[str] = set()
+    for key in ("workspace", "artifacts"):
+        value = environment.get(key)
+        if not isinstance(value, str) or not value:
+            continue
+        rel = PurePosixPath(value)
+        if rel.is_absolute():
+            continue
+        parts = [part for part in rel.parts if part != ".."]
+        if parts and parts[0] != ".":
+            names.add(parts[0])
+    return names
+
+
+def _clear_staged_tree(base_dir: Path, preserved: set[str]) -> None:
+    if not base_dir.is_dir():
+        return
+    for child in base_dir.iterdir():
+        if child.name in preserved:
+            continue
+        if child.is_dir() and not child.is_symlink():
+            shutil.rmtree(child, ignore_errors=True)
+        else:
+            child.unlink(missing_ok=True)
+
+
+def _staged_relative_paths(base_dir: Path) -> set[PurePosixPath]:
+    return {PurePosixPath(path.relative_to(base_dir).as_posix()) for path in base_dir.rglob("*") if path.is_file()}
+
+
 async def stage_fabric_spec_config_files(
     *,
     workspace: str,
@@ -127,21 +212,23 @@ def _validate_staged_size(config_files: list[ConfigFile], fileset_name: str) ->
         )
 
 
-def _validate_referenced_skill_paths(
+def validate_referenced_skill_paths(
     agent_config: dict[str, Any],
-    config_files: list[ConfigFile],
-    agent_yaml_path: str,
+    staged_paths: Collection[PurePosixPath],
 ) -> None:
+    """Ensure every ``skills.paths`` entry resolves to staged content.
+
+    *staged_paths* holds the staged files relative to the agent base directory,
+    so container ``config_files`` and an on-disk base directory validate against
+    one definition of "the skill is present".
+    """
     skills = agent_config.get("skills")
     if not isinstance(skills, dict):
         return
     paths = skills.get("paths")
     if not isinstance(paths, list) or not paths:
         return
 
-    base_dir = PurePosixPath(agent_yaml_path).parent
-    staged_paths = {PurePosixPath(config_file.path) for config_file in config_files}
-
     for skill_path in paths:
         if not isinstance(skill_path, str) or not skill_path:
             continue
@@ -150,10 +237,27 @@ def _validate_referenced_skill_paths(
             raise FabricArtifactStagingError(
                 f"Invalid skills.paths entry {skill_path!r}: must be a relative path under the agent base directory"
             )
-        expected = base_dir / rel
-        prefix = f"{expected}/"
-        matched = any(str(staged) == str(expected) or str(staged).startswith(prefix) for staged in staged_paths)
+        # "." normalizes to zero parts — the agent root itself, which packaging accepts.
+        rel_parts = tuple(part for part in rel.parts if part != ".")
+        if rel_parts:
+            matched = any(staged.parts[: len(rel_parts)] == rel_parts for staged in staged_paths)
+        else:
+            matched = bool(staged_paths)
         if not matched:
             raise FabricArtifactStagingError(
                 f"Referenced skills.paths entry {skill_path!r} was not found in staged agent spec fileset"
             )
+
+
+def _validate_referenced_skill_paths(
+    agent_config: dict[str, Any],
+    config_files: list[ConfigFile],
+    agent_yaml_path: str,
+) -> None:
+    base_dir = PurePosixPath(agent_yaml_path).parent
+    staged_paths = {
+        PurePosixPath(config_file.path).relative_to(base_dir)
+        for config_file in config_files
+        if PurePosixPath(config_file.path).is_relative_to(base_dir)
+    }
+    validate_referenced_skill_paths(agent_config, staged_paths)

plugins/nemo-agents/src/nemo_agents_plugin/runner/in_memory.py

@@ -40,6 +40,8 @@
 from nemo_agents_plugin.entities import AGENT_CONFIG_FILENAME, NEMO_AGENTS_SPEC_CONFIG_FORMAT, DeploymentMode
 from nemo_agents_plugin.fabric.gateway_credentials import platform_gateway_credential_env
 from nemo_agents_plugin.runner.backend import DeploymentInfo, LocalLog, LogLocation, NotYetAvailable, RunnerBackend
+from nemo_agents_plugin.runner.fabric_artifact_staging import stage_fabric_spec_dir
+from nemo_platform_plugin.sdk_provider import get_async_platform_sdk
 
 # Match characters not safe for filesystem paths.  Deployment names are
 # normally URL-safe identifiers, but we sanitise defensively to ensure we
@@ -228,9 +230,9 @@ async def create_deployment(
         # created_by drives on-behalf-of delegation only for container modes (via
         # the auth-proxy sidecar). Subprocess deployments run in-process on the
         # platform host and do not use the sidecar, so it does not apply here.
-        del agent, image, deployment_mode, created_by
+        del image, deployment_mode, created_by
         if config.get("config_format") == NEMO_AGENTS_SPEC_CONFIG_FORMAT:
-            return await self._create_fabric_deployment(workspace, name, config, port)
+            return await self._create_fabric_deployment(workspace, name, config, port, agent=agent)
 
         key = (workspace, name)
         config_path = await asyncio.to_thread(self._write_config, workspace, name, config)
@@ -268,12 +270,15 @@ async def _create_fabric_deployment(
         name: str,
         config: dict[str, Any],
         port: int,
+        *,
+        agent: str = "",
     ) -> DeploymentInfo:
         """Validate and start a Platform-owned Fabric-backed agent server."""
         key = (workspace, name)
         base_dir = self._fabric_base_dir_for(workspace, name)
         await asyncio.to_thread(base_dir.mkdir, parents=True, exist_ok=True)
         try:
+            await self._stage_agent_spec(workspace, agent, config, base_dir)
             config_path = await asyncio.to_thread(self._write_fabric_config, base_dir, config)
             await validate_platform_agent_config(config, base_dir=base_dir)
             log_path = self.log_path_for(workspace, name)
@@ -406,6 +411,23 @@ def _write_fabric_config(self, base_dir: Path, config: dict[str, Any]) -> Path:
         """Write a Platform-owned agent config into its deployment directory."""
         return _write_yaml_config(base_dir / AGENT_CONFIG_FILENAME, config)
 
+    async def _stage_agent_spec(
+        self,
+        workspace: str,
+        agent: str,
+        config: dict[str, Any],
+        base_dir: Path,
+    ) -> None:
+        """Deliver the agent's spec fileset (skills, MCP servers, prompts) into *base_dir*."""
+        sdk = get_async_platform_sdk(as_service="agents", internal=True) if agent else None
+        await stage_fabric_spec_dir(
+            workspace=workspace,
+            agent_name=agent,
+            agent_config=config,
+            base_dir=base_dir,
+            sdk=sdk.files if sdk else None,
+        )
+
     def _spawn(
         self,
         name: str,