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

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

@@ -1,12 +1,16 @@
 # 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 asyncio
 import logging
+import posixpath
+import shutil
 import tempfile
+from collections.abc import Collection
 from pathlib import Path, PurePosixPath
 from typing import Any, Protocol
 
@@ -38,6 +42,136 @@ 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.
+
+    ``AGENT-SPEC.md`` is dropped after download, matching the container path,
+    so both runtimes see the same tree.
+
+    The container byte cap does not apply here: it bounds ConfigMap and env
+    delivery, neither of which is in this path. What lands on the platform host
+    is bounded by ``_check_agent_root_bounds`` at CLI upload time only — a
+    fileset written straight through the files API is unbounded here. That is
+    the intended trust boundary for local subprocess mode, not an oversight.
+
+    *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.
+    """
+    preserved = _runtime_dir_names(agent_config)
+    await asyncio.to_thread(_clear_staged_tree, base_dir, preserved)
+
+    if agent_name and sdk is not None:
+        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,
+            )
+
+    staged = await asyncio.to_thread(_collect_staged_dir, base_dir, preserved)
+    validate_referenced_skill_paths(agent_config, staged)
+
+
+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(posixpath.normpath(value))
+        if rel.is_absolute():
+            continue
+        # normpath resolves "foo/../workspace" to "workspace"; a leading ".." escapes base_dir.
+        if rel.parts and rel.parts[0] != "..":
+            names.add(rel.parts[0])
+    return names
+
+
+def _clear_staged_tree(base_dir: Path, preserved: set[str]) -> None:
+    """Remove previously staged content, leaving *preserved* runtime directories.
+
+    Deletion failures are fatal: a surviving skill tree would otherwise satisfy
+    validation for a deployment that staged nothing.
+    """
+    if not base_dir.is_dir():
+        return
+    for child in base_dir.iterdir():
+        if child.name in preserved:
+            continue
+        try:
+            if child.is_dir() and not child.is_symlink():
+                shutil.rmtree(child)
+            else:
+                child.unlink(missing_ok=True)
+        except OSError as exc:
+            raise FabricArtifactStagingError(
+                f"Failed to remove stale agent spec artifact {child.name!r} from {base_dir}: {exc}"
+            ) from exc
+
+
+def _collect_staged_dir(base_dir: Path, preserved: set[str]) -> set[PurePosixPath]:
+    """Return staged files relative to *base_dir*, rejecting anything that escapes it.
+
+    ``_collect_staged_config_files`` refuses ``..`` in a staged path; this path
+    writes to the platform host's filesystem rather than an inert ConfigMap, so
+    it keeps the same guard. Symlinks are the reachable form here: the download
+    lands real files, but a link inside the tree would resolve outside it.
+    ``AGENT-SPEC.md`` is removed for parity with the container path.
+
+    *preserved* runtime directories are skipped: a running agent owns their
+    contents, symlinks included, and staging neither writes nor validates them.
+    """
+    resolved_base = base_dir.resolve()
+    staged: set[PurePosixPath] = set()
+    for path in base_dir.rglob("*"):
+        rel = path.relative_to(base_dir)
+        if rel.parts[0] in preserved:
+            continue
+        if path.is_symlink() or not path.resolve().is_relative_to(resolved_base):
+            raise FabricArtifactStagingError(
+                f"Staged agent spec path {rel.as_posix()!r} escapes the agent base directory {base_dir}"
+            )
+        if not path.is_file():
+            continue
+        if path.name == AGENT_SPEC_FILENAME:
+            path.unlink()
+            continue
+        staged.add(PurePosixPath(rel.as_posix()))
+    return staged
+
+
+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,
@@ -67,7 +201,10 @@ async def stage_fabric_spec_config_files(
                 agent_yaml_path=agent_yaml_path,
                 rewritten_agent_yaml=config_yaml,
             )
-            _validate_referenced_skill_paths(rewritten_agent_config, config_files, agent_yaml_path)
+            # Validate against what the fileset delivered, not config_files, which
+            # carries the generated agent.yaml whether or not the fileset supplied one.
+            delivered = {path for path in _staged_relative_paths(tmp_path) if path.name != AGENT_SPEC_FILENAME}
+            validate_referenced_skill_paths(rewritten_agent_config, delivered)
             _validate_staged_size(config_files, fileset_name)
             return config_files
     except (FileNotFoundError, PlatformNotFoundError, PluginClientNotFoundError) as exc:
@@ -77,11 +214,11 @@ async def stage_fabric_spec_config_files(
             fileset_name,
             exc,
         )
-        inline_only = [ConfigFile(path=agent_yaml_path, content=config_yaml)]
         # A config that references skills still needs them staged, so an absent
-        # fileset must fail here rather than at container start.
-        _validate_referenced_skill_paths(rewritten_agent_config, inline_only, agent_yaml_path)
-        return inline_only
+        # fileset must fail here rather than at container start. Nothing was
+        # staged from the fileset, so no skills.paths entry can resolve.
+        validate_referenced_skill_paths(rewritten_agent_config, set())
+        return [ConfigFile(path=agent_yaml_path, content=config_yaml)]
 
 
 def _collect_staged_config_files(
@@ -127,21 +264,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,9 +289,12 @@ 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"

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,