diff --git a/plugins/nemo-agents/openapi/openapi.yaml b/plugins/nemo-agents/openapi/openapi.yaml index b5d8f14fba..9c2b87d7c9 100644 --- a/plugins/nemo-agents/openapi/openapi.yaml +++ b/plugins/nemo-agents/openapi/openapi.yaml @@ -3932,6 +3932,20 @@ components: as ENV_VAR_NAME -> 'workspace/secret-name'. Compiled into secret-backed container env vars (never plaintext) for docker/k8s modes; ignored for subprocess. + spec_revision: + type: string + title: Spec Revision + 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. + default: '' + spec_tracked_revision: + type: string + title: Spec Tracked Revision + 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. + default: '' status: type: string enum: diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/dependencies.py b/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/dependencies.py index 510b539325..e98eeb9f90 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/dependencies.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/dependencies.py @@ -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"] diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/deployments.py b/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/deployments.py index b89a3e22e1..7ff17c03a5 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/deployments.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/api/v2/deployments.py @@ -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, diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/entities.py b/plugins/nemo-agents/src/nemo_agents_plugin/entities.py index 114eb0a8f8..483724bcdb 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/entities.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/entities.py @@ -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.", diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/runner/backend.py b/plugins/nemo-agents/src/nemo_agents_plugin/runner/backend.py index 942dc2eced..62db62df47 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/runner/backend.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/runner/backend.py @@ -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): diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/runner/controller.py b/plugins/nemo-agents/src/nemo_agents_plugin/runner/controller.py index 32eb442a72..3288ef1070 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/runner/controller.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/runner/controller.py @@ -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 diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/runner/deployments_backend.py b/plugins/nemo-agents/src/nemo_agents_plugin/runner/deployments_backend.py index 92021d6454..a1636e5712 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/runner/deployments_backend.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/runner/deployments_backend.py @@ -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() diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/runner/in_memory.py b/plugins/nemo-agents/src/nemo_agents_plugin/runner/in_memory.py index 71f937025d..d121875346 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/runner/in_memory.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/runner/in_memory.py @@ -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, diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/spec_revision.py b/plugins/nemo-agents/src/nemo_agents_plugin/spec_revision.py new file mode 100644 index 0000000000..d7d685422f --- /dev/null +++ b/plugins/nemo-agents/src/nemo_agents_plugin/spec_revision.py @@ -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 diff --git a/plugins/nemo-agents/tests/unit/test_deployments_api.py b/plugins/nemo-agents/tests/unit/test_deployments_api.py index daa60efe25..3bed753f6f 100644 --- a/plugins/nemo-agents/tests/unit/test_deployments_api.py +++ b/plugins/nemo-agents/tests/unit/test_deployments_api.py @@ -6,14 +6,16 @@ from __future__ import annotations from datetime import datetime, timezone +from types import SimpleNamespace from typing import Any from unittest.mock import AsyncMock, patch +import httpx import pytest from fastapi import FastAPI from fastapi.testclient import TestClient from nemo_agents_plugin.api.v2 import deployments as deployments_router_module -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.config import AgentsConfig from nemo_agents_plugin.entities import ( NEMO_AGENTS_SPEC_CONFIG_FORMAT, @@ -26,7 +28,9 @@ DeploymentStatus, ) from nemo_platform_plugin.auth import AuthContext +from nemo_platform_plugin.client.errors import NotFoundError as PluginClientNotFoundError from nemo_platform_plugin.entity_client import NemoEntityConflictError, NemoEntityNotFoundError +from nemo_platform_plugin.files.storage_config import GithubStorageConfig, LocalStorageConfig, StorageConfig NOW = datetime.now(timezone.utc) @@ -82,16 +86,90 @@ def _make_deployment( return deployment -def _test_client(mock_entity_client: AsyncMock) -> TestClient: +def _files_client(storage: StorageConfig | None = None) -> AsyncMock: + """A files client whose Ethos fileset is absent unless *storage* is given.""" + client = AsyncMock() + if storage is None: + client.get_fileset = AsyncMock( + side_effect=PluginClientNotFoundError(httpx.Response(404, json={"detail": "not found"})) + ) + return client + + response = AsyncMock() + response.data = lambda: SimpleNamespace(storage=storage) + client.get_fileset = AsyncMock(return_value=response) + return client + + +def _github_storage(revision: str, original_revision: str) -> GithubStorageConfig: + return GithubStorageConfig(owner="acme", repo="agents", revision=revision, original_revision=original_revision) + + +def _test_client(mock_entity_client: AsyncMock, mock_files_client: AsyncMock | None = None) -> TestClient: app = FastAPI() app.include_router( deployments_router_module.router, prefix="/apis/agents/v2/workspaces/{workspace}", ) app.dependency_overrides[get_entity_client] = lambda: mock_entity_client + app.dependency_overrides[get_files_client] = lambda: mock_files_client or _files_client() return TestClient(app, raise_server_exceptions=False) +class TestSpecRevisionSnapshot: + @staticmethod + def _create(files_client: AsyncMock) -> AgentDeployment: + mock_entity_client = AsyncMock() + mock_entity_client.get = AsyncMock(return_value=_make_agent()) + mock_entity_client.create = AsyncMock(side_effect=lambda deployment: deployment) + client = _test_client(mock_entity_client, files_client) + + resp = client.post( + "/apis/agents/v2/workspaces/default/deployments", + json={"agent": "fabric-agent", "name": "fabric-dep"}, + ) + + assert resp.status_code == 201 + return mock_entity_client.create.call_args[0][0] + + def test_records_the_revision_the_deployment_stages(self) -> None: + files_client = _files_client(_github_storage(revision="abc123", original_revision="main")) + + deployment = self._create(files_client) + + assert deployment.spec_revision == "abc123" + assert deployment.spec_tracked_revision == "main" + assert files_client.get_fileset.call_args.kwargs["name"] == "fabric-agent-ethos" + + def test_a_fileset_pinned_to_a_commit_tracks_nothing(self) -> None: + sha = "1" * 40 + deployment = self._create(_files_client(_github_storage(revision=sha, original_revision=sha))) + + assert deployment.spec_revision == sha + assert deployment.spec_tracked_revision == "" + + def test_records_nothing_for_an_agent_with_no_fileset(self) -> None: + deployment = self._create(_files_client()) + + assert deployment.spec_revision == "" + assert deployment.spec_tracked_revision == "" + + def test_records_nothing_for_a_backend_that_pins_no_revision(self) -> None: + deployment = self._create(_files_client(LocalStorageConfig(path="/data"))) + + assert deployment.spec_revision == "" + assert deployment.spec_tracked_revision == "" + + def test_an_unreadable_fileset_does_not_fail_the_deployment(self) -> None: + files_client = AsyncMock() + files_client.get_fileset = AsyncMock(side_effect=RuntimeError("files service down")) + + deployment = self._create(files_client) + + assert deployment.spec_revision == "" + assert deployment.status == "pending" + + class TestCreateDeployment: def test_create_preserves_platform_agent_config(self) -> None: mock_entity_client = AsyncMock() diff --git a/plugins/nemo-agents/tests/unit/test_runner_controller.py b/plugins/nemo-agents/tests/unit/test_runner_controller.py index 0f78bcf25a..da65786a71 100644 --- a/plugins/nemo-agents/tests/unit/test_runner_controller.py +++ b/plugins/nemo-agents/tests/unit/test_runner_controller.py @@ -32,6 +32,7 @@ ) from nemo_agents_plugin.runner.backend import DeploymentInfo from nemo_agents_plugin.runner.controller import AgentDeploymentController +from nemo_agents_plugin.spec_revision import SpecRevision from nemo_platform_plugin.auth import AuthContext from nemo_platform_plugin.entities.client import AsyncEntitiesClient from nemo_platform_plugin.entity_client import NemoEntityConflictError @@ -295,7 +296,7 @@ async def test_activity_from_current_runtime_wins_restart_reconciliation() -> No assert result is None assert current_generation.status is SessionStatus.ACTIVE - ctrl.entities.update.assert_not_called() + cast(Any, ctrl.entities.update).assert_not_called() cleanup.assert_not_called() @@ -317,7 +318,7 @@ async def test_activity_between_runtime_start_and_controller_observation_stays_a assert first_activity_at < controller_observed_at assert result is None assert current_generation.status is SessionStatus.ACTIVE - ctrl.entities.update.assert_not_called() + cast(Any, ctrl.entities.update).assert_not_called() cleanup.assert_not_called() @@ -573,6 +574,54 @@ async def test_start_deployment_writes_runtime_fields_to_entity() -> None: assert starting_key in ctrl._starting_since +@pytest.mark.asyncio +async def test_start_deployment_records_the_spec_revision_the_backend_staged() -> None: + """A restage after a fileset refresh moves the deployment, so the entity moves too.""" + ctrl, backend = _make_controller() + backend.allocate_port = MagicMock(return_value=0) + backend.create_deployment = AsyncMock( + return_value=DeploymentInfo( + name="dep-1", + status="starting", + staged_spec=SpecRevision(revision="2" * 40, tracked_revision="main"), + ) + ) + dep = AgentDeployment( + name="dep-1", + workspace="default", + agent="calc", + status="pending", + spec_revision="1" * 40, + spec_tracked_revision="main", + ) + + await ctrl._start_deployment(dep) + + assert dep.spec_revision == "2" * 40 + assert dep.spec_tracked_revision == "main" + + +@pytest.mark.asyncio +async def test_start_deployment_keeps_the_spec_revision_when_nothing_was_staged() -> None: + """A backend that stages no fileset reports nothing, which is not the same as empty.""" + ctrl, backend = _make_controller() + backend.allocate_port = MagicMock(return_value=0) + backend.create_deployment = AsyncMock(return_value=DeploymentInfo(name="dep-1", status="starting")) + dep = AgentDeployment( + name="dep-1", + workspace="default", + agent="calc", + status="pending", + spec_revision="1" * 40, + spec_tracked_revision="main", + ) + + await ctrl._start_deployment(dep) + + assert dep.spec_revision == "1" * 40 + assert dep.spec_tracked_revision == "main" + + @pytest.mark.asyncio async def test_start_deployment_forwards_image_entrypoint_mode() -> None: ctrl, backend = _make_controller() @@ -723,7 +772,7 @@ async def test_expire_session_leaves_session_before_deadline_active(expires_at: assert result is None assert session.status is SessionStatus.ACTIVE - ctrl.entities.update.assert_not_called() + cast(Any, ctrl.entities.update).assert_not_called() cleanup.assert_not_called() diff --git a/plugins/nemo-agents/tests/unit/test_spec_revision.py b/plugins/nemo-agents/tests/unit/test_spec_revision.py new file mode 100644 index 0000000000..724402bf7c --- /dev/null +++ b/plugins/nemo-agents/tests/unit/test_spec_revision.py @@ -0,0 +1,90 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Recording the revision a deployment actually staged.""" + +from __future__ import annotations + +from collections.abc import Iterator +from contextlib import contextmanager +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch + +from nemo_agents_plugin.spec_revision import SpecRevision, read_spec_revision, stage_with_spec_revision +from nemo_platform_plugin.files.storage_config import GithubStorageConfig + +FIRST_SHA = "1" * 40 +SECOND_SHA = "2" * 40 +THIRD_SHA = "3" * 40 + + +@contextmanager +def _fileset_reporting(*revisions: str) -> Iterator[None]: + """Serve each revision in turn for the agent's fileset, then the last one forever.""" + responses = [ + SimpleNamespace( + data=lambda revision=revision: SimpleNamespace( + storage=GithubStorageConfig(owner="acme", repo="agents", revision=revision, original_revision="main") + ) + ) + for revision in revisions + ] + client = AsyncMock() + client.get_fileset = AsyncMock(side_effect=lambda **_: responses.pop(0) if len(responses) > 1 else responses[0]) + + with patch("nemo_agents_plugin.spec_revision.client_from_platform", return_value=client): + yield + + +class TestStageWithSpecRevision: + async def test_records_the_revision_staging_read(self) -> None: + stage = AsyncMock(return_value="staged") + + with _fileset_reporting(FIRST_SHA): + staged, spec = await stage_with_spec_revision( + MagicMock(), workspace="default", agent_name="calc", stage=stage + ) + + assert (staged, spec) == ("staged", SpecRevision(revision=FIRST_SHA, tracked_revision="main")) + assert stage.await_count == 1 + + async def test_restages_when_a_refresh_lands_mid_stage(self) -> None: + stage = AsyncMock(return_value="staged") + + with _fileset_reporting(FIRST_SHA, SECOND_SHA, SECOND_SHA): + _, spec = await stage_with_spec_revision(MagicMock(), workspace="default", agent_name="calc", stage=stage) + + # Recording SECOND_SHA against content downloaded at FIRST_SHA is the bug; + # the second staging pass is what makes the recorded revision true. + assert spec.revision == SECOND_SHA + assert stage.await_count == 2 + + async def test_records_nothing_when_it_keeps_moving(self) -> None: + """A third revision was staged by neither pass, so none of them is the answer. + + A wrong revision is worse than no revision for a client asking whether a + deployment is stale. + """ + stage = AsyncMock(return_value="staged") + + with _fileset_reporting(FIRST_SHA, SECOND_SHA, THIRD_SHA): + _, spec = await stage_with_spec_revision(MagicMock(), workspace="default", agent_name="calc", stage=stage) + + assert spec == SpecRevision() + assert stage.await_count == 2 + + async def test_reads_nothing_without_a_files_client(self) -> None: + """The API path hands in whatever the request could adapt, including nothing.""" + assert await read_spec_revision(None, workspace="default", agent_name="calc") == SpecRevision() + + async def test_stages_anyway_when_provenance_cannot_be_read(self) -> None: + """An SDK the adapter rejects costs the deployment its revision, not its deployment.""" + stage = AsyncMock(return_value="staged") + + with patch("nemo_agents_plugin.spec_revision.client_from_platform", side_effect=TypeError("not a client")): + staged, spec = await stage_with_spec_revision( + MagicMock(), workspace="default", agent_name="calc", stage=stage + ) + + assert (staged, spec) == ("staged", SpecRevision()) + assert stage.await_count == 1