feat(agents): add AgentEnvironment / EnvironmentSpec / ComputeSpec entities

plugins/nemo-agents/src/nemo_agents_plugin/agent_config.py

@@ -4,10 +4,10 @@
 """Platform-owned agent.yaml config models for NeMo Agents.
 
 These models back Agent.config when config_format is nemo-agents-spec-v1.
-RFC122 proposes first-class environment_spec, sandbox_spec, and harness_spec
-fields on Agent; until those shapes are finalized, this config keeps those
-inputs together in the versioned Agent.config payload and can be migrated once
-the RFC122 contract lands.
+First-class environment_spec, sandbox_spec, and harness_spec fields on Agent
+are planned; until those shapes are finalized, this config keeps those inputs
+together in the versioned Agent.config payload and can be migrated once that
+contract lands.
 """
 
 from __future__ import annotations
@@ -50,6 +50,14 @@ class EnvironmentConfig(BaseModel):
     workspace: str = "./workspace"
     artifacts: str = "./artifacts"
     settings: dict[str, Any] = Field(default_factory=dict)
+    # Fabric environment mirror fields. Additive with backward-compatible
+    # defaults; populated when an AgentEnvironmentSpec is merged at deploy time and
+    # forwarded into FabricConfig.environment by the translator.
+    env: dict[str, str] = Field(default_factory=dict)
+    control_location: str | None = None
+    ownership: str | None = None
+    connection: dict[str, Any] = Field(default_factory=dict)
+    metadata: dict[str, Any] = Field(default_factory=dict)
 
 
 class TelemetryConfig(BaseModel):

plugins/nemo-agents/src/nemo_agents_plugin/api/v2/_perms.py

@@ -31,3 +31,24 @@ class DeploymentPerms(PermissionSet, namespace="agents.deployments"):
 
 class GatewayPerms(PermissionSet, namespace="agents.gateway"):
     INVOKE = perm("Invoke a deployed agent through the gateway proxy")
+
+
+class EnvironmentPerms(PermissionSet, namespace="agents.environments"):
+    CREATE = perm("Create agent environments")
+    LIST = perm("List agent environments")
+    READ = perm("Read an agent environment")
+    DELETE = perm("Delete an agent environment")
+
+
+class EnvironmentSpecPerms(PermissionSet, namespace="agents.environment-specs"):
+    CREATE = perm("Create agent environment specs")
+    LIST = perm("List agent environment specs")
+    READ = perm("Read an agent environment spec")
+    DELETE = perm("Delete an agent environment spec")
+
+
+class ComputeSpecPerms(PermissionSet, namespace="agents.compute-specs"):
+    CREATE = perm("Create agent compute specs")
+    LIST = perm("List agent compute specs")
+    READ = perm("Read an agent compute spec")
+    DELETE = perm("Delete an agent compute spec")

plugins/nemo-agents/src/nemo_agents_plugin/api/v2/deployments.py

@@ -28,8 +28,15 @@
 from nemo_agents_plugin.entities import (
     Agent,
     AgentDeployment,
+    AgentEnvironmentInline,
     is_container_deployment_mode,
 )
+from nemo_agents_plugin.environment_resolution import (
+    EnvironmentResolutionError,
+    ResolvedEnvironment,
+    merge_environment_spec_into_agent_config,
+    resolve_environment,
+)
 from nemo_agents_plugin.schema import (
     CreateDeploymentRequest,
     DeploymentFilter,
@@ -83,12 +90,27 @@ async def create_deployment(
     # Platform-owned agent specs stay strict and are translated by the runner.
     resolved_config = _resolve_deployment_config(agent, workspace=workspace)
 
-    # 4. Create the entity with status "pending"
+    # 4. Resolve and snapshot the referenced AgentEnvironment. The environment
+    # spec is merged into the resolved config (Agent-config-wins precedence); the
+    # compute spec and secret-env references are snapshotted for the container
+    # backend. Once created, a deployment is not kept in sync with the underlying
+    # environment entities.
+    resolved_environment = await _resolve_deployment_environment(
+        body.environment, workspace=workspace, entity_client=entity_client
+    )
+    resolved_config = merge_environment_spec_into_agent_config(resolved_config, resolved_environment.environment_spec)
+    env_spec = resolved_environment.environment_spec
+    resolved_secrets = dict(env_spec.secrets) if env_spec is not None else {}
+
+    # 5. Create the entity with status "pending"
     deployment = AgentDeployment(
         name=deployment_name,
         workspace=workspace,
         agent=body.agent,
         config=resolved_config,
+        environment=body.environment,
+        compute=resolved_environment.compute_spec,
+        secrets=resolved_secrets,
         status="pending",
         deployment_mode=body.deployment_mode,
         image=body.image,
@@ -120,6 +142,20 @@ def _resolve_deployment_config(agent: Agent, *, workspace: str) -> dict[str, Any
         raise HTTPException(status_code=400, detail=str(exc)) from exc
 
 
+async def _resolve_deployment_environment(
+    environment: str | AgentEnvironmentInline | None,
+    *,
+    workspace: str,
+    entity_client: NemoEntitiesClient,
+) -> ResolvedEnvironment:
+    try:
+        return await resolve_environment(environment, workspace=workspace, entity_client=entity_client)
+    except EnvironmentResolutionError as exc:
+        # 422: the request is syntactically valid but references an environment
+        # (or one of its specs) that cannot be resolved (e.g. a missing entity).
+        raise HTTPException(status_code=422, detail=str(exc)) from exc
+
+
 @router.get("/deployments", response_model=DeploymentPage, tags=["Agent Deployments"])
 @scope.read
 @path_rule(