fix(docker,jobs): scope container cleanup to owner labels and preserve pause/cancel status

docs/set-up/config-reference.mdx

@@ -493,6 +493,8 @@ models:
         model_puller_max_workers: 1
         # Number of retries when the puller fails with a transient error (IncompleteRead, ChunkedEncodingError, connection broken). Same volume is reused so partial downloads can be completed. | default: 3
         model_puller_retries: 3
+        # Additional labels copied onto model resources managed by the models controller.
+        model_labels: {}
         # BusyBox image repository used for helper containers (permissions/find/chown). | default: 'busybox'
         busybox_image: busybox
         # BusyBox image tag used for helper containers. | default: 'latest'
@@ -577,6 +579,8 @@ models:
         default_tolerations:
         # Default Kubernetes node selector for all NIM deployments. Can be overridden per-deployment via k8s_nim_operator_config. Example: {'node-type': 'gpu-node', 'zone': 'us-west1-a'}
         default_node_selector:
+        # Additional labels copied onto model resources managed by the models controller.
+        model_labels:
         # Default Kubernetes labels applied to NIMService and NIMCache resources and their child resources (e.g. pods). Merged with controller-managed labels; controller labels take precedence on conflict. Example: {'team': 'ml-platform', 'environment': 'prod'}
         default_labels:
         # Default Kubernetes annotations applied to NIMService and NIMCache resources and their child resources (e.g. pods, PVCs). Merged with controller-managed annotations; controller annotations take precedence on conflict. Example: {'prometheus.io/scrape': 'true'}

packages/nmp_testing/src/nmp/testing/docker.py

@@ -304,26 +304,38 @@ def ensure_mock_sidecar_image(docker_client: docker.DockerClient, image_name: st
 MODELS_CONTROLLER_MANAGED_LABEL = "nmp.nvidia.com/managed-by=models-controller"
 
 
-def cleanup_model_deployment_containers(docker_client: docker.DockerClient) -> int:
-    """Stop and remove all containers managed by the models controller.
+def _matches_labels(container_labels: dict[str, str], labels: dict[str, str]) -> bool:
+    """Return True when all expected Docker labels are present."""
+    return all(container_labels.get(key) == value for key, value in labels.items())
+
+
+def cleanup_model_deployment_containers(
+    docker_client: docker.DockerClient,
+    labels: dict[str, str] | None = None,
+) -> int:
+    """Stop and remove models-controller containers matching optional owner labels.
 
     Finds containers with label nmp.nvidia.com/managed-by=models-controller
-    (NIM and sidecar containers created by the Docker backend), stops and
-    removes them. Intended for integration test teardown so failed tests
-    don't leave stuck containers; use as a pytest fixture teardown.
+    (NIM, sidecar, and puller containers created by the Docker backend), stops
+    and removes them. When labels are supplied, only containers matching all
+    owner labels are removed. Intended for integration test teardown so failed
+    tests don't leave stuck containers; use as a pytest fixture teardown.
 
     Uses the same retry logic as DockerTestContext for DinD compatibility.
 
     Args:
         docker_client: Docker client to use.
+        labels: Optional labels used to scope cleanup to this test owner.
 
     Returns:
         Number of containers removed.
     """
+    labels = labels or {}
     try:
         containers = docker_client.containers.list(
             all=True,
             filters={"label": MODELS_CONTROLLER_MANAGED_LABEL},
+            ignore_removed=True,
         )
     except Exception:
         return 0
@@ -333,6 +345,9 @@ def cleanup_model_deployment_containers(docker_client: docker.DockerClient) -> i
 
     for container in containers:
         try:
+            if labels and not _matches_labels(container.labels or {}, labels):
+                continue
+
             name = container.name
 
             @retry(
@@ -500,10 +515,15 @@ def print_diagnostics(self) -> None:
         # Print relevant containers
         print("\n--- Docker PS (test containers) ---")
         try:
-            for c in self.docker_client.containers.list(all=True):
+            for c in self.docker_client.containers.list(all=True, ignore_removed=True):
                 labels = c.labels
                 if labels.get("nmp.nvidia.com/managed-by") == "models-controller":
-                    print(f"  {c.name}: {c.status}")
+                    owner = {
+                        key: value
+                        for key, value in labels.items()
+                        if key in {"nmp.nvidia.com/test-run", "nmp.nvidia.com/test-worker"}
+                    }
+                    print(f"  {c.name}: {c.status} labels={owner}")
         except Exception as e:
             print(f"Could not list containers: {e}")
 

packages/nmp_testing/tests/unit/test_docker.py

@@ -0,0 +1,54 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Unit tests for Docker testing helpers."""
+
+from unittest.mock import MagicMock
+
+from nmp.testing.docker import MODELS_CONTROLLER_MANAGED_LABEL, cleanup_model_deployment_containers
+
+
+def _container(name: str, labels: dict[str, str]) -> MagicMock:
+    container = MagicMock()
+    container.name = name
+    container.labels = labels
+    return container
+
+
+def test_cleanup_model_deployment_containers_filters_by_owner_labels():
+    """Cleanup removes only managed containers matching the supplied owner labels."""
+    owner_labels = {
+        "nmp.nvidia.com/test-run": "run-1",
+        "nmp.nvidia.com/test-worker": "gw0",
+    }
+    matching_container = _container(
+        "matching",
+        {
+            MODELS_CONTROLLER_MANAGED_LABEL: "models-controller",
+            **owner_labels,
+        },
+    )
+    other_worker_container = _container(
+        "other-worker",
+        {
+            MODELS_CONTROLLER_MANAGED_LABEL: "models-controller",
+            "nmp.nvidia.com/test-run": "run-1",
+            "nmp.nvidia.com/test-worker": "gw1",
+        },
+    )
+
+    docker_client = MagicMock()
+    docker_client.containers.list.return_value = [matching_container, other_worker_container]
+
+    removed = cleanup_model_deployment_containers(docker_client, labels=owner_labels)
+
+    assert removed == 1
+    docker_client.containers.list.assert_called_once_with(
+        all=True,
+        filters={"label": MODELS_CONTROLLER_MANAGED_LABEL},
+        ignore_removed=True,
+    )
+    matching_container.stop.assert_called_once()
+    matching_container.remove.assert_called_once_with(force=True)
+    other_worker_container.stop.assert_not_called()
+    other_worker_container.remove.assert_not_called()

services/core/inference-gateway/tests/integration/conftest.py

@@ -220,18 +220,34 @@ def docker_test_context(
 
 
 @pytest.fixture
-def models_controller_container_cleanup(docker_client: docker.DockerClient) -> Generator[None, None, None]:
+def docker_owner_labels(worker_id: str, testrun_uid: str) -> dict[str, str]:
+    """Labels used to scope Docker resources to this pytest worker/run."""
+    return {
+        "nmp.nvidia.com/test-run": testrun_uid,
+        "nmp.nvidia.com/test-worker": worker_id,
+    }
+
+
+@pytest.fixture
+def models_controller_container_cleanup(
+    docker_client: docker.DockerClient,
+    docker_owner_labels: dict[str, str],
+) -> Generator[None, None, None]:
     """Teardown: remove all containers with label nmp.nvidia.com/managed-by=models-controller.
 
     Ensures failed tests (e.g. stuck in PENDING) don't leave NIM/sidecar containers.
     Request this via controller_with_docker_and_igw; no per-test try/finally needed.
     """
     yield
-    cleanup_model_deployment_containers(docker_client)
+    cleanup_model_deployment_containers(docker_client, labels=docker_owner_labels)
 
 
 @pytest.fixture
-def docker_backend_config(worker_id: str, mock_sidecar_image: str) -> dict[str, Any]:
+def docker_backend_config(
+    worker_id: str,
+    mock_sidecar_image: str,
+    docker_owner_labels: dict[str, str],
+) -> dict[str, Any]:
     """Configuration for Docker backend in tests.
 
     Uses worker_id from pytest-xdist to allocate unique port ranges per worker.
@@ -244,6 +260,7 @@ def docker_backend_config(worker_id: str, mock_sidecar_image: str) -> dict[str,
         "models_docker_port_range_end": end_port,
         "docker_timeout": 60,
         "models_docker_host_service_name": "localhost",
+        "model_labels": docker_owner_labels,
     }
 
 

services/core/jobs/src/nmp/core/jobs/controllers/backends/kubernetes/kubernetes_job.py

@@ -311,7 +311,26 @@ def create_step_update(self, step: PlatformJobStepWithContext, job: V1Job) -> Jo
             error_details["message"] = status_details.get("message", "Job encountered an error")
         status_details["events"] = self.get_kube_job_events(job)
         task_has_error = update_all_tasks(self._nmp_sdk, self._core_v1, self.namespace, step)
-        if task_has_error:
+        teardown_lifecycle_statuses = {
+            PlatformJobStatus.PAUSING,
+            PlatformJobStatus.PAUSED,
+            PlatformJobStatus.CANCELLING,
+            PlatformJobStatus.CANCELLED,
+        }
+        # Task-level errors can appear while Kubernetes is tearing down pods for
+        # a requested pause or cancel. Preserve those user-requested lifecycle
+        # states so the dispatcher can finish transitioning to PAUSED/CANCELLED.
+        if task_has_error and status in teardown_lifecycle_statuses:
+            logger.debug(
+                "Task error observed during container teardown",
+                extra={
+                    "workspace": step.workspace,
+                    "job": step.job,
+                    "step": step.name,
+                    "status": status,
+                },
+            )
+        elif task_has_error:
             status = PlatformJobStatus.ERROR
             if "message" not in error_details:
                 error_details["message"] = "One or more tasks are in error state"

services/core/jobs/tests/controllers/test_kubernetes_backend.py

@@ -1048,6 +1048,45 @@ def test_sync_job_paused_with_errored_pods_from_sigterm(kubernetes_job, test_ste
     assert job_update.status == PlatformJobStatus.PAUSED
 
 
+def test_sync_job_paused_ignores_task_errors_from_suspend(kubernetes_job, test_step_pending):
+    """Task-level pod errors observed during suspension must not override PAUSED."""
+    mock_job_spec = MagicMock()
+    mock_job_spec.suspend = True
+
+    mock_job_status = MagicMock()
+    mock_job_status.active = None
+    mock_job_status.succeeded = None
+    mock_job_status.failed = None
+    mock_job_status.terminating = None
+    mock_job_status.completion_time = None
+
+    mock_job = MagicMock()
+    mock_job.status = mock_job_status
+    mock_job.spec = mock_job_spec
+    kubernetes_job._batch_v1.read_namespaced_job.return_value = mock_job
+
+    with (
+        patch("nmp.core.jobs.controllers.backends.kubernetes.kubernetes_job.list_pod_status") as mock_list_pod_status,
+        patch("nmp.core.jobs.controllers.backends.kubernetes.kubernetes_job.update_all_tasks") as mock_update_all_tasks,
+    ):
+        mock_list_pod_status.return_value = [
+            PodStatus(
+                task_id="test-task",
+                name="test-pod",
+                errors={"test-container": 137},
+                completed=set(),
+                active=set(),
+                waiting={},
+                phase="Failed",
+            )
+        ]
+        mock_update_all_tasks.return_value = True
+
+        job_update = kubernetes_job.sync(test_step_pending)
+
+    assert job_update.status == PlatformJobStatus.PAUSED
+
+
 def test_sync_job_pausing_with_errored_pods_from_sigterm(kubernetes_job, test_step_pending):
     """Test that a suspended job with both running and errored pods reports PAUSING, not ERROR.
 
@@ -1136,6 +1175,45 @@ def test_sync_job_cancelling_with_errored_pods(kubernetes_job, test_step_cancell
     assert job_update.status == PlatformJobStatus.CANCELLED
 
 
+def test_sync_job_cancelled_ignores_task_errors_from_termination(kubernetes_job, test_step_cancelling):
+    """Task-level pod errors observed during termination must not override CANCELLED."""
+    mock_job_spec = MagicMock()
+    mock_job_spec.suspend = False
+
+    mock_job_status = MagicMock()
+    mock_job_status.active = None
+    mock_job_status.succeeded = None
+    mock_job_status.failed = None
+    mock_job_status.terminating = None
+    mock_job_status.completion_time = None
+
+    mock_job = MagicMock()
+    mock_job.status = mock_job_status
+    mock_job.spec = mock_job_spec
+    kubernetes_job._batch_v1.read_namespaced_job.return_value = mock_job
+
+    with (
+        patch("nmp.core.jobs.controllers.backends.kubernetes.kubernetes_job.list_pod_status") as mock_list_pod_status,
+        patch("nmp.core.jobs.controllers.backends.kubernetes.kubernetes_job.update_all_tasks") as mock_update_all_tasks,
+    ):
+        mock_list_pod_status.return_value = [
+            PodStatus(
+                task_id="test-task",
+                name="test-pod",
+                errors={"test-container": 137},
+                completed=set(),
+                active=set(),
+                waiting={},
+                phase="Failed",
+            )
+        ]
+        mock_update_all_tasks.return_value = True
+
+        job_update = kubernetes_job.sync(test_step_cancelling)
+
+    assert job_update.status == PlatformJobStatus.CANCELLED
+
+
 def test_sync_job_cancelling(kubernetes_job, test_step_cancelling):
     """Test syncing a cancelling job that isn't ready to be cancelled."""
     # Mock cancelling job status

services/core/models/src/nmp/core/models/controllers/backends/docker/backend.py

@@ -45,6 +45,11 @@
 logger = getLogger(__name__)
 
 
+def _matches_labels(container_labels: dict[str, str], expected_labels: dict[str, str]) -> bool:
+    """Return True when all expected Docker labels are present."""
+    return all(container_labels.get(key) == value for key, value in expected_labels.items())
+
+
 class DockerServiceBackend(ServiceBackend):
     """Docker-based backend for managing model deployments.
 
@@ -439,14 +444,18 @@ async def list_managed_deployment_names(self) -> list[str]:
                 self._reconciler.list_containers,
                 all=True,
                 filters={"label": f"{MODEL_MANAGED_BY_LABEL}={MODEL_MANAGED_BY_MODELS_CONTROLLER}"},
+                ignore_removed=True,
             )
         except Exception as e:
             logger.warning(f"Failed to list managed containers for orphan reconciliation: {e}")
             return []
 
         seen: set[str] = set()
+        owner_labels = self._backend_config.model_labels
         for container in containers:
             labels = container.labels or {}
+            if owner_labels and not _matches_labels(labels, owner_labels):
+                continue
             ws = labels.get("nmp.nvidia.com/deployment-workspace")
             n = labels.get("nmp.nvidia.com/deployment-name")
             if ws and n:

services/core/models/src/nmp/core/models/controllers/backends/docker/config.py

@@ -174,6 +174,11 @@ class DockerBackendConfig(BaseModel):
         "ChunkedEncodingError, connection broken). Same volume is reused so partial downloads can be completed.",
     )
 
+    model_labels: dict[str, str] = Field(
+        default_factory=dict,
+        description=("Additional labels copied onto model resources managed by the models controller. "),
+    )
+
     busybox_image: str = Field(
         default="busybox",
         description="BusyBox image repository used for helper containers (permissions/find/chown).",