NVIDIA-NeMo · benmccown · Sep 9, 2026 · Sep 8, 2026 · Sep 9, 2026 · Sep 9, 2026
@@ -577,13 +577,6 @@ basePlatformConfig: |
           enabled: true
           k8s_executor: local-k8s
           default_executor: local-k8s
-          # Platform-default pod annotations applied to every k8s model
-          # deployment/job (all engines). The Istio native-sidecar annotation
-          # makes a mesh-injected istio-proxy terminate when a Job's main
-          # container exits, so the weight-puller Job completes while keeping
-          # mesh mTLS egress (inject: "false" is NOT viable in-mesh).
-          default_pod_annotations:
-            sidecar.istio.io/nativeSidecar: "true"
           # Bypass the image `nemo` ENTRYPOINT; invoke the adapters module directly.
           # Avoids PermissionError when the sidecar writes instance state under $HOME.
           lora_sidecar_command:

@@ -139,6 +139,7 @@ async def create_deployment(
                 backend_config=backend_config,
                 config=config,
                 executor_image_pull_secrets=self._executor_config.image_pull_secrets,
+                executor_defaults=self._executor_config.to_k8s_defaults(),
                 secret_env=secret_env,
                 auth_context=auth_context,
                 workload_delegation_store=self._workload_delegations,
@@ -154,6 +155,7 @@ async def create_deployment(
             backend_config=backend_config,
             config=config,
             executor_image_pull_secrets=self._executor_config.image_pull_secrets,
+            executor_defaults=self._executor_config.to_k8s_defaults(),
             secret_env=secret_env,
             auth_context=auth_context,
             workload_delegation_store=self._workload_delegations,

@@ -155,6 +155,24 @@ def build_container_spec(
     return k8s.client.V1Container(**kwargs)
 
 
+@dataclass(frozen=True)
+class ExecutorK8sDefaults:
+    """Executor-level k8s defaults applied to every workload the executor renders.
+
+    These are the base layer shared by ALL deployments-plugin consumers (models,
+    agents, ...). A per-entity ``K8sDeploymentConfig`` value overrides the default:
+    annotations merge key-wise (per-entity key wins); node_selector / tolerations /
+    affinity / topology_spread_constraints are applied only when the entity leaves
+    them unset (per-entity wins wholesale).
+    """
+
+    pod_annotations: dict[str, str] = field(default_factory=dict)
+    node_selector: dict[str, str] = field(default_factory=dict)
+    tolerations: list[dict[str, Any]] = field(default_factory=list)
+    affinity: dict[str, Any] = field(default_factory=dict)
+    topology_spread_constraints: list[dict[str, Any]] = field(default_factory=list)
+
+
 @dataclass(frozen=True)
 class CompiledWorkload:
     """Kubernetes objects derived from a DeploymentConfig."""
@@ -531,6 +549,7 @@ def compile_workload(
     k8s_config: K8sDeploymentConfig | None,
     pod_restart_policy: RestartPolicy,
     executor_image_pull_secrets: list[ImagePullSecret] | None = None,
+    executor_defaults: ExecutorK8sDefaults | None = None,
     secret_env: dict[str, str] | None = None,
 ) -> CompiledWorkload:
     """Compile pod spec kwargs and optional ConfigMap/Secret for a Job or Deployment.
@@ -618,7 +637,32 @@ def compile_workload(
     if effective_service_account_name:
         pod_spec_kwargs["service_account_name"] = effective_service_account_name
 
-    pod_annotations = dict(k8s_config.pod_annotations) if k8s_config is not None else {}
+    # Apply executor-level defaults as the BASE layer, shared by every consumer
+    # (models, agents, ...). A per-entity K8sDeploymentConfig value (set above)
+    # wins: node_selector / tolerations / affinity / topology_spread are applied
+    # only when the entity left them unset; annotations merge key-wise below.
+    entity_pod_annotations = dict(k8s_config.pod_annotations) if k8s_config is not None else {}
+    if executor_defaults is not None:
+        if executor_defaults.node_selector and "node_selector" not in pod_spec_kwargs:
+            pod_spec_kwargs["node_selector"] = dict(executor_defaults.node_selector)
+        if executor_defaults.tolerations and "tolerations" not in pod_spec_kwargs:
+            default_tolerations = build_tolerations(
+                [Toleration.model_validate(item) for item in executor_defaults.tolerations if item]
+            )
+            if default_tolerations:
+                pod_spec_kwargs["tolerations"] = default_tolerations
+        if executor_defaults.affinity and "affinity" not in pod_spec_kwargs:
+            default_affinity = _deserialize_k8s(executor_defaults.affinity, "V1Affinity")
+            if default_affinity is not None:
+                pod_spec_kwargs["affinity"] = default_affinity
+        if executor_defaults.topology_spread_constraints and "topology_spread_constraints" not in pod_spec_kwargs:
+            default_tsc = build_topology_spread_constraints(executor_defaults.topology_spread_constraints)
+            if default_tsc:
+                pod_spec_kwargs["topology_spread_constraints"] = default_tsc
+        # Annotations merge key-wise: executor default first, per-entity wins.
+        pod_annotations = {**executor_defaults.pod_annotations, **entity_pod_annotations}
+    else:
+        pod_annotations = entity_pod_annotations
 
     return CompiledWorkload(
         pod_spec_kwargs=pod_spec_kwargs,

@@ -7,7 +7,9 @@
 
 import os
 import re
+from typing import Any
 
+from nemo_deployments_plugin.backends.k8s.compiler import ExecutorK8sDefaults
 from nemo_platform_plugin.config import ImagePullSecret
 from pydantic import BaseModel, Field, field_validator
 
@@ -50,6 +52,47 @@ class K8sExecutorConfig(BaseModel):
         default_factory=list,
         description="Image pull secrets merged with platform image_pull_secrets on every pod.",
     )
+    default_pod_annotations: dict[str, str] = Field(
+        default_factory=dict,
+        description=(
+            "Executor-level default pod annotations stamped onto every Job/Deployment pod this "
+            "executor renders (all consumers: models, agents, ...). Merged key-wise with per-entity "
+            "backend_config.k8s.podAnnotations, where the per-entity value wins for a shared key. "
+            "Ships the Istio native-sidecar annotation so a mesh-injected proxy terminates when a "
+            "Job's main container exits."
+        ),
+    )
+    default_node_selector: dict[str, str] = Field(
+        default_factory=dict,
+        description=(
+            "Executor-level default nodeSelector applied to every Job/Deployment pod this executor "
+            "renders when the entity does not set backend_config.k8s.nodeSelector."
+        ),
+    )
+    default_tolerations: list[dict[str, str | int]] = Field(
+        default_factory=list,
+        description=(
+            "Executor-level default pod tolerations applied to every Job/Deployment pod this executor "
+            "renders when the entity does not set backend_config.k8s.tolerations. Each entry is a raw "
+            "Kubernetes toleration object."
+        ),
+    )
+    default_affinity: dict[str, Any] = Field(
+        default_factory=dict,
+        description=(
+            "Executor-level default pod affinity applied to every Job/Deployment pod this executor "
+            "renders when the entity does not set backend_config.k8s.affinity. Raw Kubernetes affinity "
+            "object (nodeAffinity / podAffinity / podAntiAffinity)."
+        ),
+    )
+    default_topology_spread_constraints: list[dict[str, Any]] = Field(
+        default_factory=list,
+        description=(
+            "Executor-level default pod topology spread constraints applied to every Job/Deployment pod "
+            "this executor renders when the entity does not set backend_config.k8s.topologySpreadConstraints. "
+            "Each entry is a raw Kubernetes topologySpreadConstraint object."
+        ),
+    )
 
     @field_validator("default_namespace")
     @classmethod
@@ -73,3 +116,18 @@ def effective_namespace(self) -> str:
         if pod_namespace:
             return pod_namespace
         return _FALLBACK_NAMESPACE
+
+    def to_k8s_defaults(self) -> ExecutorK8sDefaults:
+        """Bundle the executor-level pod defaults for the workload compiler.
+
+        Shared by every deployments-plugin consumer (models, agents, ...): the
+        compiler applies these as the base layer, with per-entity
+        ``backend_config.k8s`` values overriding them.
+        """
+        return ExecutorK8sDefaults(
+            pod_annotations=dict(self.default_pod_annotations),
+            node_selector=dict(self.default_node_selector),
+            tolerations=[dict(item) for item in self.default_tolerations],
+            affinity=dict(self.default_affinity),
+            topology_spread_constraints=[dict(item) for item in self.default_topology_spread_constraints],
+        )
@@ -21,6 +21,7 @@
 from nemo_deployments_plugin.backends.k8s.compiler import (
     CompiledWorkload,
     DeploymentConfigError,
+    ExecutorK8sDefaults,
     compile_workload,
     create_configmap,
     create_secret,
@@ -101,6 +102,7 @@ def build_deployment_body(
     deployment_name: str,
     k8s_config: K8sDeploymentConfig | None,
     executor_image_pull_secrets: list | None = None,
+    executor_defaults: ExecutorK8sDefaults | None = None,
     secret_env: dict[str, str] | None = None,
 ) -> BuiltDeployment:
     """Build an ``apps/v1.Deployment`` for create and its compiled workload."""
@@ -115,6 +117,7 @@ def build_deployment_body(
         k8s_config=k8s_config,
         pod_restart_policy="Always",
         executor_image_pull_secrets=executor_image_pull_secrets,
+        executor_defaults=executor_defaults,
         secret_env=secret_env,
     )
     deployment = k8s.client.V1Deployment(
@@ -331,6 +334,7 @@ async def create_deployment(
     backend_config: dict[str, Any],
     config: DeploymentConfig,
     executor_image_pull_secrets: list | None = None,
+    executor_defaults: ExecutorK8sDefaults | None = None,
     secret_env: dict[str, str] | None = None,
     auth_context: AuthContext | None = None,
     workload_delegation_store: WorkloadDelegationStore | None = None,
@@ -359,6 +363,7 @@ async def create_deployment(
             deployment_name=name,
             k8s_config=k8s_config,
             executor_image_pull_secrets=executor_image_pull_secrets,
+            executor_defaults=executor_defaults,
             secret_env=secret_env,
         )
         deployment_body = built.deployment

@@ -18,6 +18,7 @@
 from nemo_deployments_plugin.backends.k8s.compiler import (
     CompiledWorkload,
     DeploymentConfigError,
+    ExecutorK8sDefaults,
     compile_workload,
     create_configmap,
     create_secret,
@@ -147,6 +148,7 @@ def build_job_body(
     deployment_name: str,
     k8s_config: K8sDeploymentConfig | None,
     executor_image_pull_secrets: list | None = None,
+    executor_defaults: ExecutorK8sDefaults | None = None,
     secret_env: dict[str, str] | None = None,
 ) -> BuiltJob:
     """Build a ``batch/v1.Job`` for create."""
@@ -159,6 +161,7 @@ def build_job_body(
         k8s_config=k8s_config,
         pod_restart_policy=config.restart_policy,
         executor_image_pull_secrets=executor_image_pull_secrets,
+        executor_defaults=executor_defaults,
         secret_env=secret_env,
     )
     job = k8s.client.V1Job(
@@ -249,6 +252,7 @@ async def create_job(
     backend_config: dict[str, Any],
     config: DeploymentConfig,
     executor_image_pull_secrets: list | None = None,
+    executor_defaults: ExecutorK8sDefaults | None = None,
     secret_env: dict[str, str] | None = None,
     auth_context: AuthContext | None = None,
     workload_delegation_store: WorkloadDelegationStore | None = None,
@@ -277,6 +281,7 @@ async def create_job(
             deployment_name=name,
             k8s_config=k8s_config,
             executor_image_pull_secrets=executor_image_pull_secrets,
+            executor_defaults=executor_defaults,
             secret_env=secret_env,
         )
         body = built.job

@@ -30,6 +30,31 @@ def test_default_namespace_rejects_invalid_dns_label() -> None:
         K8sExecutorConfig(default_namespace="X")
 
 
+def test_to_k8s_defaults_bundles_executor_pod_defaults() -> None:
+    config = K8sExecutorConfig(
+        default_pod_annotations={"sidecar.istio.io/nativeSidecar": "true"},
+        default_node_selector={"gpu": "a100"},
+        default_tolerations=[{"key": "gpu", "operator": "Equal", "value": "true", "effect": "NoSchedule"}],
+        default_affinity={"nodeAffinity": {}},
+        default_topology_spread_constraints=[{"maxSkew": 1, "topologyKey": "kubernetes.io/hostname"}],
+    )
+    defaults = config.to_k8s_defaults()
+    assert defaults.pod_annotations == {"sidecar.istio.io/nativeSidecar": "true"}
+    assert defaults.node_selector == {"gpu": "a100"}
+    assert defaults.tolerations[0]["key"] == "gpu"
+    assert defaults.affinity == {"nodeAffinity": {}}
+    assert defaults.topology_spread_constraints[0]["topologyKey"] == "kubernetes.io/hostname"
+
+
+def test_to_k8s_defaults_empty_by_default() -> None:
+    defaults = K8sExecutorConfig().to_k8s_defaults()
+    assert defaults.pod_annotations == {}
+    assert defaults.node_selector == {}
+    assert defaults.tolerations == []
+    assert defaults.affinity == {}
+    assert defaults.topology_spread_constraints == []
+
+
 def test_effective_namespace_prefers_explicit_config(monkeypatch: pytest.MonkeyPatch) -> None:
     # An explicit config value wins even when POD_NAMESPACE is set.
     monkeypatch.setenv("POD_NAMESPACE", "pod-ns")

@@ -11,6 +11,7 @@
 from kubernetes.client import ApiClient
 from nemo_deployments_plugin.backends.k8s.compiler import (
     DeploymentConfigError,
+    ExecutorK8sDefaults,
     _build_probe,
     build_configmap_body,
     build_env_vars,
@@ -263,6 +264,99 @@ def test_compile_carries_pod_annotations() -> None:
     assert "annotations" not in compiled.pod_spec_kwargs
 
 
+def test_compile_applies_executor_defaults() -> None:
+    # Executor-level defaults (shared by every consumer: models, agents, ...) land
+    # on a workload with no per-entity k8s config.
+    config = sample_always_config()
+    executor_defaults = ExecutorK8sDefaults(
+        pod_annotations={"sidecar.istio.io/nativeSidecar": "true"},
+        node_selector={"gpu": "a100"},
+        tolerations=[{"key": "gpu", "operator": "Equal", "value": "true", "effect": "NoSchedule"}],
+        affinity={
+            "nodeAffinity": {
+                "requiredDuringSchedulingIgnoredDuringExecution": {
+                    "nodeSelectorTerms": [{"matchExpressions": [{"key": "gpu", "operator": "In", "values": ["a100"]}]}]
+                }
+            }
+        },
+        topology_spread_constraints=[
+            {"maxSkew": 1, "topologyKey": "kubernetes.io/hostname", "whenUnsatisfiable": "DoNotSchedule"}
+        ],
+    )
+    compiled = compile_workload(
+        config=config,
+        workspace="default",
+        deployment_name="task",
+        labels={"managed-by": "nemo-deployments"},
+        k8s_config=None,
+        pod_restart_policy="Always",
+        executor_defaults=executor_defaults,
+    )
+    pod_spec = _serialized(compiled.pod_spec_kwargs)
+    assert compiled.pod_annotations == {"sidecar.istio.io/nativeSidecar": "true"}
+    assert pod_spec["node_selector"] == {"gpu": "a100"}
+    assert pod_spec["tolerations"][0]["key"] == "gpu"
+    assert compiled.pod_spec_kwargs["affinity"].node_affinity is not None
+    assert pod_spec["topology_spread_constraints"][0]["topologyKey"] == "kubernetes.io/hostname"
+
+
+def test_compile_per_entity_wins_over_executor_defaults() -> None:
+    # A per-entity K8sDeploymentConfig overrides the executor default: annotations
+    # merge key-wise (entity key wins), and node_selector / tolerations are applied
+    # from the entity wholesale (executor default not additionally applied).
+    config = sample_always_config()
+    k8s_config = K8sDeploymentConfig.model_validate(
+        {
+            "podAnnotations": {"sidecar.istio.io/nativeSidecar": "false", "team": "a"},
+            "nodeSelector": {"zone": "us-west1-a"},
+            "tolerations": [{"key": "entity", "operator": "Exists"}],
+        }
+    )
+    executor_defaults = ExecutorK8sDefaults(
+        pod_annotations={"sidecar.istio.io/nativeSidecar": "true", "platform": "nmp"},
+        node_selector={"gpu": "a100"},
+        tolerations=[{"key": "platform", "operator": "Exists"}],
+    )
+    compiled = compile_workload(
+        config=config,
+        workspace="default",
+        deployment_name="task",
+        labels={"managed-by": "nemo-deployments"},
+        k8s_config=k8s_config,
+        pod_restart_policy="Always",
+        executor_defaults=executor_defaults,
+    )
+    pod_spec = _serialized(compiled.pod_spec_kwargs)
+    # Annotations: entity value wins for the shared key; non-conflicting keys from
+    # both sides are retained.
+    assert compiled.pod_annotations == {
+        "sidecar.istio.io/nativeSidecar": "false",
+        "team": "a",
+        "platform": "nmp",
+    }
+    # node_selector / tolerations: entity wins wholesale.
+    assert pod_spec["node_selector"] == {"zone": "us-west1-a"}
+    assert [t["key"] for t in pod_spec["tolerations"]] == ["entity"]
+
+
+def test_compile_empty_executor_defaults_are_noop() -> None:
+    config = sample_config(restart_policy="Never")
+    compiled = compile_workload(
+        config=config,
+        workspace="default",
+        deployment_name="task",
+        labels={"managed-by": "nemo-deployments"},
+        k8s_config=None,
+        pod_restart_policy="Never",
+        executor_defaults=ExecutorK8sDefaults(),
+    )
+    assert compiled.pod_annotations == {}
+    assert "node_selector" not in compiled.pod_spec_kwargs
+    assert "tolerations" not in compiled.pod_spec_kwargs
+    assert "affinity" not in compiled.pod_spec_kwargs
+    assert "topology_spread_constraints" not in compiled.pod_spec_kwargs
+
+
 def test_compile_pod_annotations_default_empty_without_k8s_config() -> None:
     config = sample_config(restart_policy="Never")
     compiled = compile_workload(