fix(jobs): refuse deleting active jobs

e2e/auditor/test_audit_job.py

@@ -21,6 +21,7 @@
 import pytest
 from nemo_platform import NeMoPlatform
 from nmp.testing import add_mock_provider, short_unique_name
+from nmp.testing.e2e import cleanup_platform_job
 
 from e2e.auditor.utils import minimal_audit_config, unique_name
 
@@ -56,10 +57,13 @@ def _wait_for_audit_job(sdk: NeMoPlatform, job_name: str, workspace: str) -> str
 
 
 def _cleanup_audit_job(sdk: NeMoPlatform, job_name: str, workspace: str) -> None:
-    with suppress(Exception):
-        sdk.jobs.cancel(name=job_name, workspace=workspace)
-    with suppress(Exception):
-        sdk.jobs.delete(name=job_name, workspace=workspace)
+    cleanup_platform_job(
+        sdk,
+        job_name,
+        workspace,
+        timeout=AUDIT_JOB_TIMEOUT_SECONDS,
+        poll_interval=AUDIT_JOB_POLL_INTERVAL_SECONDS,
+    )
 
 
 def _add_mock_provider_or_skip(sdk: NeMoPlatform, workspace: str, name: str) -> str:

e2e/test_anonymizer_plugin.py

@@ -39,6 +39,7 @@
 from nemo_platform_plugin.files.client import FilesClient
 from nemo_platform_plugin.files.types import CreateFilesetRequest
 from nmp.testing import MockProviderResponse, add_mock_provider, short_unique_name
+from nmp.testing.e2e import cleanup_platform_job
 
 pytestmark = [
     pytest.mark.container_only,
@@ -323,10 +324,13 @@ def _wait_for_anonymizer_job(job: AnonymizerJobResource, *, timeout_seconds: flo
 
 
 def _cleanup_anonymizer_job(sdk: NeMoPlatform, job_name: str) -> None:
-    with suppress(Exception):
-        sdk.jobs.cancel(name=job_name, workspace=sdk.workspace)
-    with suppress(Exception):
-        sdk.jobs.delete(name=job_name, workspace=sdk.workspace)
+    cleanup_platform_job(
+        sdk,
+        job_name,
+        str(sdk.workspace),
+        timeout=ANONYMIZER_JOB_TIMEOUT_SECONDS,
+        poll_interval=ANONYMIZER_POLL_INTERVAL_SECONDS,
+    )
 
 
 @pytest.fixture(scope="module")

e2e/test_evaluator_plugin.py

@@ -49,7 +49,7 @@
 from nemo_platform import APIConnectionError, APIStatusError, NeMoPlatform
 from nemo_platform.types.inference import ModelProvider
 from nmp.testing import add_mock_provider, short_unique_name, wait_for_model_entity
-from nmp.testing.e2e import wait_for_platform_job
+from nmp.testing.e2e import cleanup_platform_job, wait_for_platform_job
 from nmp.testing.utils import ensure_passthrough_virtual_model
 
 pytestmark = [
@@ -286,10 +286,13 @@ def _create_ready_mock_model(
 
 
 def _cleanup_evaluator_job(sdk: NeMoPlatform, job_name: str) -> None:
-    with suppress(Exception):
-        sdk.jobs.cancel(name=job_name, workspace=sdk.workspace)
-    with suppress(Exception):
-        sdk.jobs.delete(name=job_name, workspace=sdk.workspace)
+    cleanup_platform_job(
+        sdk,
+        job_name,
+        str(sdk.workspace),
+        timeout=EVALUATOR_JOB_TIMEOUT_SECONDS,
+        poll_interval=EVALUATOR_POLL_INTERVAL_SECONDS,
+    )
 
 
 def _wait_for_evaluator_job(job: EvaluatorJobResource) -> None:

e2e/test_safe_synthesizer.py

@@ -345,14 +345,21 @@ def _wait_for_job_absent(
 
 
 def _delete_nss_job(sdk: NeMoPlatform, workspace: str, name: str, *, verify: bool = True) -> None:
-    response = sdk._client.delete(
-        _nss_url(sdk, workspace, f"jobs/{name}"),
-        headers=_string_headers(sdk),
-        timeout=60.0,
-    )
-    if response.status_code not in {200, 202, 204, 404}:
-        response.raise_for_status()
-    if verify:
+    deadline = time.monotonic() + DELETE_VERIFY_TIMEOUT_SECONDS
+    while True:
+        response = sdk._client.delete(
+            _nss_url(sdk, workspace, f"jobs/{name}"),
+            headers=_string_headers(sdk),
+            timeout=60.0,
+        )
+        if response.status_code in {200, 202, 204, 404}:
+            break
+        if response.status_code != 409:
+            response.raise_for_status()
+        if time.monotonic() >= deadline:
+            raise TimeoutError(f"Timed out deleting Safe Synthesizer job {name!r}; last response={response.text}")
+        time.sleep(2.0)
+    if verify and response.status_code != 404:
         _wait_for_job_absent(sdk, workspace, name)
 
 
@@ -583,9 +590,42 @@ def create(prefix: str, data_source: str, config: dict[str, Any]) -> dict[str, A
     try:
         yield create
     finally:
+        cleanup_errors: list[Exception] = []
         for job_name in reversed(job_names):
-            _cancel_nss_job(sdk, workspace, job_name)
-            _delete_nss_job(sdk, workspace, job_name)
+            try:
+                cancel_response = _cancel_nss_job(sdk, workspace, job_name)
+                if cancel_response is not None:
+                    _wait_for_status(
+                        sdk,
+                        workspace,
+                        job_name,
+                        timeout_seconds=SMOKE_JOB_TIMEOUT_SECONDS,
+                        poll_interval_seconds=2.0,
+                    )
+            except Exception as exc:
+                cleanup_errors.append(exc)
+                _capture_nss_debug_artifacts(
+                    sdk,
+                    workspace,
+                    job_name,
+                    "cleanup cancel/wait failed",
+                    history=[],
+                    error=exc,
+                )
+            try:
+                _delete_nss_job(sdk, workspace, job_name)
+            except Exception as exc:
+                cleanup_errors.append(exc)
+                _capture_nss_debug_artifacts(
+                    sdk,
+                    workspace,
+                    job_name,
+                    "cleanup delete failed",
+                    history=[],
+                    error=exc,
+                )
+        if cleanup_errors:
+            raise ExceptionGroup("Safe Synthesizer job cleanup failed", cleanup_errors)
 
 
 def test_safe_synthesizer_api_health(sdk: NeMoPlatform, workspace: str) -> None:

packages/nemo_platform_plugin/src/nemo_platform_plugin/jobs/api_factory.py

@@ -51,6 +51,8 @@
 from nemo_platform_plugin.api.parsed_filter import ParsedFilter, make_filter_dep
 from nemo_platform_plugin.authz import AuthzScope, CallerKind, path_rule
 from nemo_platform_plugin.client.adapter import client_from_platform
+from nemo_platform_plugin.client.errors import NemoHTTPError
+from nemo_platform_plugin.client.types import RetryPolicy
 from nemo_platform_plugin.dependencies import get_entity_client, get_sdk_client
 from nemo_platform_plugin.entities import EntityClient
 from nemo_platform_plugin.jobs.client import AsyncJobsClient
@@ -1083,14 +1085,22 @@ async def get_job_status(
         @router.delete(
             "/jobs/{name}",
             status_code=status.HTTP_204_NO_CONTENT,
+            responses={
+                status.HTTP_404_NOT_FOUND: {"description": "Job not Found"},
+                status.HTTP_409_CONFLICT: {"description": "Job is not in a terminal state"},
+            },
         )
         async def delete_job(
             workspace: str,
             name: str,
             sdk: AsyncNeMoPlatform = Depends(get_sdk_client),
         ) -> None:
             f"""Delete a job by name for the {service_name} microservice."""
-            await client_from_platform(sdk, AsyncJobsClient).delete_job(name=name, workspace=workspace)
+            try:
+                jobs_client = client_from_platform(sdk, AsyncJobsClient).with_retry(RetryPolicy(max_retries=0))
+                await jobs_client.delete_job(name=name, workspace=workspace)
+            except NemoHTTPError as exc:
+                raise HTTPException(status_code=exc.status_code, detail=exc.detail) from exc
             return None
 
         @router.post(

packages/nmp_common/src/nmp/common/docker/gpu_pool.py

@@ -17,7 +17,30 @@
 class GPUAllocationError(Exception):
     """Raised when GPU allocation fails due to insufficient resources."""
 
-    pass
+    def __init__(
+        self,
+        message: str,
+        *,
+        requested: int | None = None,
+        available: int | None = None,
+        total: int | None = None,
+    ) -> None:
+        super().__init__(message)
+        self.message = message
+        self.requested = requested
+        self.available = available
+        self.total = total
+
+    @property
+    def is_transient_capacity_exhaustion(self) -> bool:
+        """Whether a retry could succeed after another workload releases GPUs."""
+        return (
+            self.requested is not None
+            and self.available is not None
+            and self.total is not None
+            and self.requested > 0
+            and self.available < self.requested <= self.total
+        )
 
 
 @dataclass
@@ -92,9 +115,13 @@ def allocate_gpu(self, workload_id: str, num_requested: int = 1) -> list[int]:
             available_gpus = {gpu for gpu, workload in self.gpu_to_workload_id.items() if workload is None}
 
             if len(available_gpus) < num_requested:
+                available_count = len(available_gpus)
                 raise GPUAllocationError(
                     f"Not enough GPUs available. Requested {num_requested}, "
-                    f"available {len(available_gpus)} out of {self.num_reserved_gpus} total."
+                    f"available {available_count} out of {self.num_reserved_gpus} total.",
+                    requested=num_requested,
+                    available=available_count,
+                    total=self.num_reserved_gpus,
                 )
             gpu_ids = []
             for _ in range(num_requested):

packages/nmp_common/tests/api_factory/test_api_factory.py

@@ -355,6 +355,7 @@ def _job_routes_app():
 
     mock_sdk = MagicMock()
     mock_jobs = MagicMock()
+    mock_jobs.with_retry.return_value = mock_jobs
     router = job_route_factory(
         service_name="test_service",
         job_type="TestJob",

packages/nmp_common/tests/docker/test_gpu_pool.py

@@ -71,14 +71,17 @@ def test_allocate_sequential_workloads(self):
         assert set(gpu_ids_1).isdisjoint(set(gpu_ids_2))
 
     @pytest.mark.parametrize(
-        "pool_size,pre_allocate,num_requested,expected_error_fragment",
+        "pool_size,pre_allocate,num_requested,expected_error_fragment,expected_transient",
         [
-            pytest.param(2, 1, 2, "Requested 2", id="insufficient_remaining"),
-            pytest.param(1, 1, 1, "Requested 1", id="none_available"),
-            pytest.param(0, 0, 1, "Requested 1", id="empty_pool"),
+            pytest.param(2, 1, 2, "Requested 2", True, id="insufficient_remaining"),
+            pytest.param(1, 1, 1, "Requested 1", True, id="none_available"),
+            pytest.param(1, 0, 2, "Requested 2", False, id="request_exceeds_total"),
+            pytest.param(0, 0, 1, "Requested 1", False, id="empty_pool"),
         ],
     )
-    def test_allocate_raises_when_insufficient(self, pool_size, pre_allocate, num_requested, expected_error_fragment):
+    def test_allocate_raises_when_insufficient(
+        self, pool_size, pre_allocate, num_requested, expected_error_fragment, expected_transient
+    ):
         """Test that allocation raises GPUAllocationError when not enough GPUs available."""
         pool = DockerGPUPool(reserved_gpu_device_ids=list(range(pool_size)))
         if pre_allocate > 0:
@@ -88,6 +91,10 @@ def test_allocate_raises_when_insufficient(self, pool_size, pre_allocate, num_re
             pool.allocate_gpu("workload-new", num_requested=num_requested)
 
         assert expected_error_fragment in str(exc_info.value)
+        assert exc_info.value.requested == num_requested
+        assert exc_info.value.available == pool_size - pre_allocate
+        assert exc_info.value.total == pool_size
+        assert exc_info.value.is_transient_capacity_exhaustion is expected_transient
 
     @pytest.mark.parametrize(
         "invalid_value",
@@ -107,6 +114,7 @@ def test_allocate_raises_on_invalid_num_requested(self, invalid_value):
 
         assert "Invalid GPU request" in str(exc_info.value)
         assert "Must be a positive integer" in str(exc_info.value)
+        assert exc_info.value.is_transient_capacity_exhaustion is False
 
 
 class TestDockerGPUPoolRelease:

packages/nmp_testing/src/nmp/testing/e2e/__init__.py

@@ -43,7 +43,7 @@
     load_config,
 )
 from .docker import Docker
-from .jobs import wait_for_job_completion, wait_for_job_logs, wait_for_platform_job
+from .jobs import cleanup_platform_job, wait_for_job_completion, wait_for_job_logs, wait_for_platform_job
 from .kubernetes import Kubernetes
 
 __all__ = [
@@ -59,6 +59,7 @@
     "discover_configs",
     "infer_backend",
     "load_config",
+    "cleanup_platform_job",
     "wait_for_job_completion",
     "wait_for_job_logs",
     "wait_for_platform_job",