NVIDIA-NeMo · marcusds · Sep 11, 2026 · Sep 11, 2026
@@ -7,6 +7,21 @@
 a single location.
 """
 
+from fastapi import Depends
+from nemo_agents_plugin.spec_revision import files_client_for
+from nemo_platform import AsyncNeMoPlatform
+from nemo_platform_plugin.dependencies import get_sdk_client
 from nemo_platform_plugin.entity_client import get_entity_client
+from nemo_platform_plugin.files.client import AsyncFilesClient
 
-__all__ = ["get_entity_client"]
+
+def get_files_client(sdk: AsyncNeMoPlatform = Depends(get_sdk_client)) -> AsyncFilesClient | None:
+    """Provide a Files service client sharing the request's SDK transport.
+
+    None when the SDK cannot be adapted. What this client is used for — recording
+    which revision a deployment staged — must never fail the deployment.
+    """
+    return files_client_for(sdk)
+
+
+__all__ = ["get_entity_client", "get_files_client"]
@@ -23,7 +23,7 @@
 from fastapi import APIRouter, Depends, HTTPException, Query, Request
 from nemo_agents_plugin.agent_config_formats import AgentConfigFormatError, resolve_agent_config_for_deployment
 from nemo_agents_plugin.api.v2._perms import DeploymentPerms
-from nemo_agents_plugin.api.v2.dependencies import get_entity_client
+from nemo_agents_plugin.api.v2.dependencies import get_entity_client, get_files_client
 from nemo_agents_plugin.authz import scope
 from nemo_agents_plugin.config import AgentsConfig
 from nemo_agents_plugin.entities import (
@@ -49,10 +49,12 @@
     DeploymentFilter,
     DeploymentPage,
 )
+from nemo_agents_plugin.spec_revision import read_spec_revision
 from nemo_platform_plugin.api.filters import make_filter_obj_dep
 from nemo_platform_plugin.auth import current_auth_context
 from nemo_platform_plugin.authz import CallerKind, path_rule
 from nemo_platform_plugin.entity_client import NemoEntitiesClient, NemoEntityConflictError, NemoEntityNotFoundError
+from nemo_platform_plugin.files.client import AsyncFilesClient
 from nemo_platform_plugin.schema import PaginationData
 
 logger = logging.getLogger(__name__)
@@ -74,6 +76,7 @@ async def create_deployment(
     body: CreateDeploymentRequest,
     request: Request,
     entity_client: NemoEntitiesClient = Depends(get_entity_client),
+    files_client: AsyncFilesClient = Depends(get_files_client),
 ) -> AgentDeployment:
     """Create a new deployment for an existing agent.
 
@@ -136,7 +139,11 @@ async def create_deployment(
     )
     merged = _merge_environment(resolved_config, resolved_environment.environment_spec)
 
-    # 5. Create the entity with status "pending"
+    # 5. Record the spec fileset revision this deployment starts from. The runner
+    # rewrites it with whatever it actually stages.
+    spec = await read_spec_revision(files_client, workspace=workspace, agent_name=body.agent)
+
+    # 6. Create the entity with status "pending"
     deployment = AgentDeployment(
         name=deployment_name,
         workspace=workspace,
@@ -145,6 +152,8 @@ async def create_deployment(
         environment=body.environment,
         compute=resolved_environment.compute_spec,
         secrets=merged.secrets,
+        spec_revision=spec.revision,
+        spec_tracked_revision=spec.tracked_revision,
         status="pending",
         deployment_mode=body.deployment_mode,
         image=body.image,

@@ -442,6 +442,22 @@ class AgentDeployment(NemoEntity, entity_type="agent_deployment"):
             "vars (never plaintext) for docker/k8s modes; ignored for subprocess."
         ),
     )
+    # Written when the deployment is created and rewritten every time the runner
+    # restages the fileset, so it names the revision the deployment is serving.
+    spec_revision: str = Field(
+        default="",
+        description=(
+            "Revision of the agent spec fileset this deployment last staged, for source-backed "
+            "filesets (e.g. a GitHub commit SHA). Empty when the fileset has no revision."
+        ),
+    )
+    spec_tracked_revision: str = Field(
+        default="",
+        description=(
+            "Mutable ref the spec fileset tracked when this deployment last staged it (e.g. 'main'). "
+            "Empty when the fileset was pinned to an immutable id."
+        ),
+    )
     status: DeploymentStatus = Field(
         default="pending",
         description="Lifecycle status: pending | starting | running | failed | deleting.",

@@ -18,6 +18,7 @@
 from typing import Any, Literal
 
 from nemo_agents_plugin.entities import ComputeResources, DeploymentMode, DeploymentStatus, Endpoint
+from nemo_agents_plugin.spec_revision import SpecRevision
 from nemo_platform_plugin.auth import AuthContext
 
 
@@ -68,6 +69,13 @@ class DeploymentInfo:
     """Absolute path to the subprocess log file (empty if not applicable)."""
     extra: dict[str, Any] = field(default_factory=dict)
     """Backend-specific metadata (e.g. container ID for Docker)."""
+    staged_spec: SpecRevision | None = None
+    """What the spec fileset resolved to when this deployment staged it.
+
+    None when the backend staged nothing, which is not the same as staging a
+    fileset that pins no revision — the controller leaves the recorded revision
+    alone in the first case and overwrites it in the second.
+    """
 
 
 class RunnerBackend(ABC):

@@ -572,6 +572,11 @@ async def _start_deployment(self, dep: AgentDeployment) -> None:
             return
 
         spawn_ms = (time.perf_counter() - t0) * 1000
+        if info.staged_spec is not None:
+            # Staging reads the fileset as it is now, so a deployment restarted after
+            # a refresh reports the revision it actually restaged from.
+            dep.spec_revision = info.staged_spec.revision
+            dep.spec_tracked_revision = info.staged_spec.tracked_revision
         dep.status = info.status
         dep.port = info.port
         dep.pid = info.pid

@@ -38,6 +38,7 @@
     FabricArtifactStagingError,
     stage_fabric_ethos_config_files,
 )
+from nemo_agents_plugin.spec_revision import SpecRevision, stage_with_spec_revision
 from nemo_agents_plugin.telemetry.intake_export import (
     configure_intake_atif_export,
     supports_intake_atif_export,
@@ -693,16 +694,22 @@ async def create_deployment(
             deployment_labels["nemo.agents/runtime"] = "fabric"
 
         staged_config_files: list[ConfigFile] | None = None
+        staged_spec: SpecRevision | None = None
         if is_fabric and agent:
             agent_yaml_path = _fabric_config_mount_path(self._config.config_mount_path)
+            sdk = get_async_platform_sdk(as_service="agents", internal=True)
             try:
-                sdk = get_async_platform_sdk(as_service="agents", internal=True)
-                staged_config_files = await stage_fabric_ethos_config_files(
+                staged_config_files, staged_spec = await stage_with_spec_revision(
+                    sdk,
                     workspace=workspace,
                     agent_name=agent,
-                    rewritten_agent_config=config,
-                    agent_yaml_path=agent_yaml_path,
-                    sdk=sdk.files,
+                    stage=lambda: stage_fabric_ethos_config_files(
+                        workspace=workspace,
+                        agent_name=agent,
+                        rewritten_agent_config=config,
+                        agent_yaml_path=agent_yaml_path,
+                        sdk=sdk.files,
+                    ),
                 )
             except FabricArtifactStagingError as exc:
                 logger.error("Refusing to deploy Fabric agent %r: %s", name, exc)
@@ -766,7 +773,7 @@ async def create_deployment(
             resolved_image,
             deployment_mode,
         )
-        return DeploymentInfo(name=name, status="starting", endpoint="", endpoints=[])
+        return DeploymentInfo(name=name, status="starting", endpoint="", endpoints=[], staged_spec=staged_spec)
 
     async def get_deployment_status(self, workspace: str, name: str) -> DeploymentInfo | None:
         entities = self._entity_client()

@@ -46,6 +46,7 @@
 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_ethos_dir
+from nemo_agents_plugin.spec_revision import SpecRevision, stage_with_spec_revision
 from nemo_platform_plugin.auth import AuthContext
 from nemo_platform_plugin.sdk_provider import get_async_platform_sdk
 
@@ -291,7 +292,7 @@ async def _create_fabric_deployment(
         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_ethos(workspace, agent, config, base_dir)
+            staged_spec = await self._stage_ethos(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)
@@ -316,6 +317,7 @@ async def _create_fabric_deployment(
             endpoint=f"http://127.0.0.1:{port}",
             log_path=str(log_path),
             extra={"base_dir": str(base_dir)},
+            staged_spec=staged_spec,
         )
         self._processes[key] = proc
         self._deployments[key] = info
@@ -430,16 +432,30 @@ async def _stage_ethos(
         agent: str,
         config: dict[str, Any],
         base_dir: Path,
-    ) -> None:
-        """Deliver the agent's Ethos fileset into *base_dir*."""
+    ) -> SpecRevision | None:
+        """Deliver the agent's Ethos fileset into *base_dir* and report what it staged."""
         sdk = get_async_platform_sdk(as_service="agents", internal=True) if agent else None
-        await stage_fabric_ethos_dir(
+
+        async def _stage() -> None:
+            await stage_fabric_ethos_dir(
+                workspace=workspace,
+                agent_name=agent,
+                agent_config=config,
+                base_dir=base_dir,
+                sdk=sdk.files if sdk else None,
+            )
+
+        if sdk is None:
+            await _stage()
+            return None
+
+        _, spec = await stage_with_spec_revision(
+            sdk,
             workspace=workspace,
             agent_name=agent,
-            agent_config=config,
-            base_dir=base_dir,
-            sdk=sdk.files if sdk else None,
+            stage=_stage,
         )
+        return spec
 
     def _spawn(
         self,

@@ -0,0 +1,118 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Read the revision an agent's spec fileset is pinned to."""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
+from typing import TypeVar
+
+from nemo_agents_plugin.entities import ethos_fileset_name
+from nemo_platform import AsyncNeMoPlatform
+from nemo_platform_plugin.client.adapter import client_from_platform
+from nemo_platform_plugin.client.errors import NotFoundError as PluginClientNotFoundError
+from nemo_platform_plugin.files.client import AsyncFilesClient
+from nemo_platform_plugin.log_utils import sanitize_for_log
+
+logger = logging.getLogger(__name__)
+
+T = TypeVar("T")
+
+
+@dataclass(frozen=True)
+class SpecRevision:
+    """The revision an agent's spec fileset resolved to, and the ref it tracks."""
+
+    revision: str = ""
+    tracked_revision: str = ""
+
+
+async def read_spec_revision(files_client: AsyncFilesClient | None, *, workspace: str, agent_name: str) -> SpecRevision:
+    """Return what the agent's spec fileset is pinned to right now.
+
+    Both fields are empty when the fileset is absent or its backend pins nothing —
+    an agent that deploys from its inline config alone is the normal case, not an
+    error. Nothing here may fail a deployment: this is a record of what was staged,
+    and the runner reports a fileset it cannot read.
+    """
+    if files_client is None:
+        return SpecRevision()
+
+    fileset_name = ethos_fileset_name(agent_name)
+    try:
+        response = await files_client.get_fileset(workspace=workspace, name=fileset_name)
+        storage = response.data().storage
+    except PluginClientNotFoundError:
+        return SpecRevision()
+    except Exception:
+        logger.warning(
+            "Could not read fileset %s/%s for deployment provenance",
+            sanitize_for_log(workspace),
+            sanitize_for_log(fileset_name),
+            exc_info=True,
+        )
+        return SpecRevision()
+
+    return SpecRevision(revision=storage.pinned_revision, tracked_revision=storage.tracked_revision or "")
+
+
+def files_client_for(sdk: AsyncNeMoPlatform) -> AsyncFilesClient | None:
+    """Adapt the platform SDK, or report that provenance cannot be read through it.
+
+    Failing to build the client is the same kind of event as an unreadable
+    fileset: it costs the deployment its recorded revision, not its deployment.
+    """
+    try:
+        return client_from_platform(sdk, AsyncFilesClient)
+    except Exception:
+        logger.warning("Could not build a files client for deployment provenance", exc_info=True)
+        return None
+
+
+async def stage_with_spec_revision(
+    sdk: AsyncNeMoPlatform,
+    *,
+    workspace: str,
+    agent_name: str,
+    stage: Callable[[], Awaitable[T]],
+) -> tuple[T, SpecRevision]:
+    """Run *stage*, and report the revision the content it staged came from.
+
+    The files API resolves a fileset's revision per request, so reading the
+    revision after the download would record whatever a refresh landing in between
+    left behind rather than what was staged. Restage once when the revision moves,
+    which is the closest thing to an atomic read the download path allows.
+    """
+    files_client = files_client_for(sdk)
+    if files_client is None:
+        return await stage(), SpecRevision()
+
+    before = await read_spec_revision(files_client, workspace=workspace, agent_name=agent_name)
+    staged = await stage()
+    after = await read_spec_revision(files_client, workspace=workspace, agent_name=agent_name)
+    if after == before:
+        return staged, after
+
+    logger.info(
+        "Spec fileset %s/%s moved from %r to %r while staging; staging it again",
+        sanitize_for_log(workspace),
+        sanitize_for_log(ethos_fileset_name(agent_name)),
+        before.revision,
+        after.revision,
+    )
+    staged = await stage()
+    settled = await read_spec_revision(files_client, workspace=workspace, agent_name=agent_name)
+    if settled != after:
+        # `settled` is a third revision that no staging pass produced. Recording it
+        # would be the failure this function exists to prevent, and a wrong revision
+        # is worse than none for a client asking whether a deployment is stale.
+        logger.warning(
+            "Spec fileset %s/%s moved again while staging; recording no revision for it",
+            sanitize_for_log(workspace),
+            sanitize_for_log(ethos_fileset_name(agent_name)),
+        )
+        return staged, SpecRevision()
+    return staged, settled