fix(jobs): refuse deleting active jobs

e2e/test_safe_synthesizer.py

@@ -350,6 +350,22 @@ def _delete_nss_job(sdk: NeMoPlatform, workspace: str, name: str, *, verify: boo
         headers=_string_headers(sdk),
         timeout=60.0,
     )
+    if response.status_code == 409:
+        try:
+            _wait_for_status(
+                sdk,
+                workspace,
+                name,
+                timeout_seconds=SMOKE_JOB_TIMEOUT_SECONDS,
+                poll_interval_seconds=2.0,
+            )
+        except Exception:
+            response.raise_for_status()
+        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:
@@ -584,7 +600,15 @@ def create(prefix: str, data_source: str, config: dict[str, Any]) -> dict[str, A
         yield create
     finally:
         for job_name in reversed(job_names):
-            _cancel_nss_job(sdk, workspace, job_name)
+            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,
+                )
             _delete_nss_job(sdk, workspace, job_name)
 
 

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

@@ -51,6 +51,7 @@
 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.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 +1084,21 @@ 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:
+                await client_from_platform(sdk, AsyncJobsClient).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/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: