From 16bd1dfdde3768581d3e21ba8120505b31c46aee Mon Sep 17 00:00:00 2001
From: eva-1729 <eva.maitland@stfc.ac.uk>
Date: Fri, 24 Apr 2026 11:35:19 +0100
Subject: [PATCH 01/29] Add cleanup to K8s resources, and retry for transient
 errors

---
 job_creator/jobcreator/job_creator.py | 334 ++++++++++++++------------
 1 file changed, 186 insertions(+), 148 deletions(-)

diff --git a/job_creator/jobcreator/job_creator.py b/job_creator/jobcreator/job_creator.py
index ced77bd..c55dc49 100644
--- a/job_creator/jobcreator/job_creator.py
+++ b/job_creator/jobcreator/job_creator.py
@@ -5,6 +5,8 @@
 from typing import Any
 
 from kubernetes import client  # type: ignore[import-untyped]
+from kubernetes.client.exceptions import ApiException
+from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception
 
 from jobcreator.utils import load_kubernetes_config, logger
 
@@ -123,6 +125,20 @@ def _setup_extras_pv(job_name: str, secret_namespace: str, manila_share_id: str,
     return pv_name
 
 
+def _is_retryable_k8s_error(exc):
+    """Only retry on transient K8s API errors, not on conflicts or client errors."""
+    if isinstance(exc, ApiException):
+        return exc.status in (429, 500, 502, 503, 504)
+    # Also retry on connection-level errors
+    return isinstance(exc, (ConnectionError, TimeoutError))
+
+
+@retry(
+        retry=retry_if_exception(_is_retryable_k8s_error),
+        stop=stop_after_attempt(3),
+        wait=wait_exponential(multiplier=1, min=1, max=10),
+        reraise=True,
+)
 def _setup_ceph_pv(
     pv_name: str,
     ceph_creds_k8s_secret_name: str,
@@ -299,168 +315,190 @@ def spawn_job(  # noqa: PLR0913
         pv_names = []
         pvc_names = []
 
-        # Setup Archive PV and PVC
-        archive_pv_name = f"{job_name}-archive-pv-smb"
-        _setup_smb_pv(
-            archive_pv_name,
-            "archive-creds",
-            job_namespace,
-            "//isisdatar55.isis.cclrc.ac.uk/inst$/",
-            ["noserverino", "_netdev", "vers=2.1"],
-        )
-        pv_names.append(archive_pv_name)
-
-        archive_pvc_name = f"{job_name}-archive-pvc"
-        _setup_pvc(archive_pvc_name, archive_pv_name, job_namespace)
-        pvc_names.append(archive_pvc_name)
-
-        # Setup Extras PV and PVC
-        extras_pv_name = _setup_extras_pv(
-            job_name=job_name,
-            secret_namespace=job_namespace,
-            manila_share_id=manila_share_id,
-            manila_share_access_id=manila_share_access_id,
-        )
-        pv_names.append(extras_pv_name)
-
-        extras_pvc_name = f"{job_name}-extras-pvc"
-        _setup_pvc(extras_pvc_name, extras_pv_name, job_namespace)
-        pvc_names.append(extras_pvc_name)
-
-        # Setup ceph PV and PVC
-        if not self.dev_mode:
-            ceph_pv_name = f"{job_name}-ceph-pv"
-            (
-                _setup_ceph_pv(
-                    ceph_pv_name,
-                    ceph_creds_k8s_secret_name,
-                    ceph_creds_k8s_namespace,
-                    cluster_id,
-                    fs_name,
-                    ceph_mount_path,
-                ),
+        try:
+            # Setup Archive PV and PVC
+            archive_pv_name = f"{job_name}-archive-pv-smb"
+            _setup_smb_pv(
+                archive_pv_name,
+                "archive-creds",
+                job_namespace,
+                "//isisdatar55.isis.cclrc.ac.uk/inst$/",
+                ["noserverino", "_netdev", "vers=2.1"],
             )
-            pv_names.append(ceph_pv_name)
+            pv_names.append(archive_pv_name)
+
+            archive_pvc_name = f"{job_name}-archive-pvc"
+            _setup_pvc(archive_pvc_name, archive_pv_name, job_namespace)
+            pvc_names.append(archive_pvc_name)
+
+            # Setup Extras PV and PVC
+            extras_pv_name = _setup_extras_pv(
+                job_name=job_name,
+                secret_namespace=job_namespace,
+                manila_share_id=manila_share_id,
+                manila_share_access_id=manila_share_access_id,
+            )
+            pv_names.append(extras_pv_name)
+
+            extras_pvc_name = f"{job_name}-extras-pvc"
+            _setup_pvc(extras_pvc_name, extras_pv_name, job_namespace)
+            pvc_names.append(extras_pvc_name)
+
+            # Setup ceph PV and PVC
+            if not self.dev_mode:
+                ceph_pv_name = f"{job_name}-ceph-pv"
+                (
+                    _setup_ceph_pv(
+                        ceph_pv_name,
+                        ceph_creds_k8s_secret_name,
+                        ceph_creds_k8s_namespace,
+                        cluster_id,
+                        fs_name,
+                        ceph_mount_path,
+                    ),
+                )
+                pv_names.append(ceph_pv_name)
+
+                ceph_pvc_name = f"{job_name}-ceph-pvc"
+                _setup_pvc(ceph_pvc_name, ceph_pv_name, job_namespace, access_mode="ReadWriteMany")
+                pvc_names.append(ceph_pvc_name)
+
+                ceph_volume = client.V1Volume(
+                    name="ceph-mount",
+                    persistent_volume_claim=client.V1PersistentVolumeClaimVolumeSource(
+                        claim_name=ceph_pvc_name,
+                        read_only=False,
+                    ),
+                )
+            else:
+                ceph_volume = client.V1Volume(
+                    name="ceph-mount",
+                    empty_dir=client.V1EmptyDirVolumeSource(size_limit="100Gi"),
+                )
 
-            ceph_pvc_name = f"{job_name}-ceph-pvc"
-            _setup_pvc(ceph_pvc_name, ceph_pv_name, job_namespace, access_mode="ReadWriteMany")
-            pvc_names.append(ceph_pvc_name)
+            # Create the Job
+            logger.info("Spawning job: %s", job_name)
 
-            ceph_volume = client.V1Volume(
-                name="ceph-mount",
-                persistent_volume_claim=client.V1PersistentVolumeClaimVolumeSource(
-                    claim_name=ceph_pvc_name,
-                    read_only=False,
+            volumes = [
+                client.V1Volume(
+                    name="archive-mount",
+                    persistent_volume_claim=client.V1PersistentVolumeClaimVolumeSource(
+                        claim_name=archive_pvc_name,
+                        read_only=True,
+                    ),
+                ),
+                ceph_volume,
+                client.V1Volume(
+                    name="extras-mount",
+                    persistent_volume_claim=client.V1PersistentVolumeClaimVolumeSource(
+                        claim_name=extras_pvc_name,
+                        read_only=True,
+                    ),
                 ),
+            ]
+            volumes_mounts = [
+                client.V1VolumeMount(name="archive-mount", mount_path="/archive"),
+                client.V1VolumeMount(name="ceph-mount", mount_path="/output"),
+                client.V1VolumeMount(name="extras-mount", mount_path="/extras"),
+            ]
+            # Setup special PVs and add them to the volume mounts
+            if "imat" in special_pvs:
+                _setup_imat_pv_and_pvcs(job_name, job_namespace, pv_names, pvc_names)
+                imat_pvc_source = client.V1PersistentVolumeClaimVolumeSource(
+                    claim_name=f"{job_name}-ndximat-pvc", read_only=True
+                )
+                volumes.append(client.V1Volume(name="imat-mount", persistent_volume_claim=imat_pvc_source))
+                volumes_mounts.append(client.V1VolumeMount(name="imat-mount", mount_path="/imat"))
+                # Because imat is special and uses mantid imaging to load large .tiff files, we need to ensure the /dev/shm
+                # is larger than 64mb. We do however have a soft-ish limit of around 32GiB on the size of datasets when
+                # doing this.
+                volumes.append(
+                    client.V1Volume(
+                        name="dev-shm", empty_dir=client.V1EmptyDirVolumeSource(size_limit="32Gi", medium="Memory")
+                    )
+                )
+                volumes_mounts.append(client.V1VolumeMount(name="dev-shm", mount_path="/dev/shm"))  # noqa: S108
+
+            main_container = client.V1Container(
+                name=job_name,
+                image=runner_image,
+                args=[script],
+                env=[client.V1EnvVar(name="PYTHONUNBUFFERED", value="1")],
+                volume_mounts=volumes_mounts,
             )
-        else:
-            ceph_volume = client.V1Volume(
-                name="ceph-mount",
-                empty_dir=client.V1EmptyDirVolumeSource(size_limit="100Gi"),
+
+            watcher_container = client.V1Container(
+                name="job-watcher",
+                image=f"ghcr.io/fiaisis/jobwatcher@sha256:{self.watcher_sha}",
+                env=[
+                    client.V1EnvVar(name="FIA_API_HOST", value=fia_api_host),
+                    client.V1EnvVar(name="FIA_API_API_KEY", value=fia_api_api_key),
+                    client.V1EnvVar(name="MAX_TIME_TO_COMPLETE_JOB", value=str(max_time_to_complete_job)),
+                    client.V1EnvVar(name="CONTAINER_NAME", value=job_name),
+                    client.V1EnvVar(name="JOB_NAME", value=job_name),
+                    client.V1EnvVar(name="POD_NAME", value=job_name),
+                ],
             )
 
-        # Create the Job
-        logger.info("Spawning job: %s", job_name)
+            affinity = _generate_affinities(node_affinity_dict=affinity)
+            tolerations = _generate_tolerations_from_taints(taints)
 
-        volumes = [
-            client.V1Volume(
-                name="archive-mount",
-                persistent_volume_claim=client.V1PersistentVolumeClaimVolumeSource(
-                    claim_name=archive_pvc_name,
-                    read_only=True,
-                ),
-            ),
-            ceph_volume,
-            client.V1Volume(
-                name="extras-mount",
-                persistent_volume_claim=client.V1PersistentVolumeClaimVolumeSource(
-                    claim_name=extras_pvc_name,
-                    read_only=True,
-                ),
-            ),
-        ]
-        volumes_mounts = [
-            client.V1VolumeMount(name="archive-mount", mount_path="/archive"),
-            client.V1VolumeMount(name="ceph-mount", mount_path="/output"),
-            client.V1VolumeMount(name="extras-mount", mount_path="/extras"),
-        ]
-        # Setup special PVs and add them to the volume mounts
-        if "imat" in special_pvs:
-            _setup_imat_pv_and_pvcs(job_name, job_namespace, pv_names, pvc_names)
-            imat_pvc_source = client.V1PersistentVolumeClaimVolumeSource(
-                claim_name=f"{job_name}-ndximat-pvc", read_only=True
-            )
-            volumes.append(client.V1Volume(name="imat-mount", persistent_volume_claim=imat_pvc_source))
-            volumes_mounts.append(client.V1VolumeMount(name="imat-mount", mount_path="/imat"))
-            # Because imat is special and uses mantid imaging to load large .tiff files, we need to ensure the /dev/shm
-            # is larger than 64mb. We do however have a soft-ish limit of around 32GiB on the size of datasets when
-            # doing this.
-            volumes.append(
-                client.V1Volume(
-                    name="dev-shm", empty_dir=client.V1EmptyDirVolumeSource(size_limit="32Gi", medium="Memory")
-                )
+            pod_spec = client.V1PodSpec(
+                affinity=affinity,
+                service_account_name="jobwatcher",
+                containers=[main_container, watcher_container],
+                restart_policy="Never",
+                tolerations=tolerations,
+                volumes=volumes,
             )
-            volumes_mounts.append(client.V1VolumeMount(name="dev-shm", mount_path="/dev/shm"))  # noqa: S108
-
-        main_container = client.V1Container(
-            name=job_name,
-            image=runner_image,
-            args=[script],
-            env=[client.V1EnvVar(name="PYTHONUNBUFFERED", value="1")],
-            volume_mounts=volumes_mounts,
-        )
 
-        watcher_container = client.V1Container(
-            name="job-watcher",
-            image=f"ghcr.io/fiaisis/jobwatcher@sha256:{self.watcher_sha}",
-            env=[
-                client.V1EnvVar(name="FIA_API_HOST", value=fia_api_host),
-                client.V1EnvVar(name="FIA_API_API_KEY", value=fia_api_api_key),
-                client.V1EnvVar(name="MAX_TIME_TO_COMPLETE_JOB", value=str(max_time_to_complete_job)),
-                client.V1EnvVar(name="CONTAINER_NAME", value=job_name),
-                client.V1EnvVar(name="JOB_NAME", value=job_name),
-                client.V1EnvVar(name="POD_NAME", value=job_name),
-            ],
-        )
+            pod_metadata = client.V1ObjectMeta(
+                labels={"reduce.isis.cclrc.ac.uk/job-source": "automated-reduction"},
+            )
 
-        affinity = _generate_affinities(node_affinity_dict=affinity)
-        tolerations = _generate_tolerations_from_taints(taints)
+            template = client.V1PodTemplateSpec(spec=pod_spec, metadata=pod_metadata)
 
-        pod_spec = client.V1PodSpec(
-            affinity=affinity,
-            service_account_name="jobwatcher",
-            containers=[main_container, watcher_container],
-            restart_policy="Never",
-            tolerations=tolerations,
-            volumes=volumes,
-        )
+            spec = client.V1JobSpec(
+                template=template,
+                backoff_limit=0,
+                ttl_seconds_after_finished=21600,  # 6 hours
+            )
 
-        pod_metadata = client.V1ObjectMeta(
-            labels={"reduce.isis.cclrc.ac.uk/job-source": "automated-reduction"},
-        )
+            job_metadata = client.V1ObjectMeta(
+                name=job_name,
+                annotations={
+                    "job-id": str(job_id),
+                    "pvs": str(pv_names),
+                    "pvcs": str(pvc_names),
+                    "kubectl.kubernetes.io/default-container": main_container.name,
+                },
+            )
 
-        template = client.V1PodTemplateSpec(spec=pod_spec, metadata=pod_metadata)
+            job = client.V1Job(
+                api_version="batch/v1",
+                kind="Job",
+                metadata=job_metadata,
+                spec=spec,
+            )
+            client.BatchV1Api().create_namespaced_job(namespace=job_namespace, body=job)
 
-        spec = client.V1JobSpec(
-            template=template,
-            backoff_limit=0,
-            ttl_seconds_after_finished=21600,  # 6 hours
-        )
+        except Exception:
+            logger.exception("Failed to spawn job %s, cleaning up resources", job_name)
+            self._cleanup_resources(pv_names, pvc_names, job_namespace)
+            raise
 
-        job_metadata = client.V1ObjectMeta(
-            name=job_name,
-            annotations={
-                "job-id": str(job_id),
-                "pvs": str(pv_names),
-                "pvcs": str(pvc_names),
-                "kubectl.kubernetes.io/default-container": main_container.name,
-            },
-        )
 
-        job = client.V1Job(
-            api_version="batch/v1",
-            kind="Job",
-            metadata=job_metadata,
-            spec=spec,
-        )
-        client.BatchV1Api().create_namespaced_job(namespace=job_namespace, body=job)
+    def _cleanup_resources(self, pv_names, pvc_names, namespace):
+        """Best-effort cleanup of created K8s resources."""
+        for pvc_name in pvc_names:
+            try:
+                client.CoreV1Api().delete_namespaced_persistent_valume_claim(
+                    name=pvc_name, namespace=namespace
+                )
+            except client.ApiException:
+                logger.warning("Failed to cleanup PVC: %s", pvc_name)
+        for pv_name in pv_names:
+            try:
+                client.CoreV1Api().delete_persistent_volume(name=pv_name)
+            except client.ApiException:
+                logger.warning("Failed to cleanup PV: %s", pv_name)

From fcd81c8f90b5b36c05d294b5a831b624b3749080 Mon Sep 17 00:00:00 2001
From: eva-1729 <eva.maitland@stfc.ac.uk>
Date: Fri, 24 Apr 2026 11:38:11 +0100
Subject: [PATCH 02/29] Add tenacity to pyproject.toml

---
 job_creator/pyproject.toml | 1 +
 1 file changed, 1 insertion(+)

diff --git a/job_creator/pyproject.toml b/job_creator/pyproject.toml
index b57b2ef..6144b9e 100644
--- a/job_creator/pyproject.toml
+++ b/job_creator/pyproject.toml
@@ -6,6 +6,7 @@ requires-python = ">= 3.11"
 dependencies = [
     "kubernetes==35.0.0",
     "pika==1.3.2",
+    "tenacity=9.1.4",
 ]
 
 [project.urls]

From 8b505ab8d55be9e06ff4ba79f7a0fe3e12436470 Mon Sep 17 00:00:00 2001
From: eva-1729 <eva.maitland@stfc.ac.uk>
Date: Fri, 24 Apr 2026 11:39:45 +0100
Subject: [PATCH 03/29] fix typo

---
 job_creator/pyproject.toml | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/job_creator/pyproject.toml b/job_creator/pyproject.toml
index 6144b9e..51f7965 100644
--- a/job_creator/pyproject.toml
+++ b/job_creator/pyproject.toml
@@ -6,7 +6,7 @@ requires-python = ">= 3.11"
 dependencies = [
     "kubernetes==35.0.0",
     "pika==1.3.2",
-    "tenacity=9.1.4",
+    "tenacity==9.1.4",
 ]
 
 [project.urls]

From 4d7b28be1909bc82f219a1e2d400b8d8ef31a931 Mon Sep 17 00:00:00 2001
From: eva-1729 <eva.maitland@stfc.ac.uk>
Date: Fri, 24 Apr 2026 11:46:02 +0100
Subject: [PATCH 04/29] ruff formatting

---
 job_creator/jobcreator/job_creator.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/job_creator/jobcreator/job_creator.py b/job_creator/jobcreator/job_creator.py
index c55dc49..e76f680 100644
--- a/job_creator/jobcreator/job_creator.py
+++ b/job_creator/jobcreator/job_creator.py
@@ -409,9 +409,9 @@ def spawn_job(  # noqa: PLR0913
                 )
                 volumes.append(client.V1Volume(name="imat-mount", persistent_volume_claim=imat_pvc_source))
                 volumes_mounts.append(client.V1VolumeMount(name="imat-mount", mount_path="/imat"))
-                # Because imat is special and uses mantid imaging to load large .tiff files, we need to ensure the /dev/shm
-                # is larger than 64mb. We do however have a soft-ish limit of around 32GiB on the size of datasets when
-                # doing this.
+                # Because imat is special and uses mantid imaging to load large .tiff files, we need to 
+                # ensure the /dev/shm is larger than 64mb. We do however have a soft-ish limit of around 
+                # 32GiB on the size of datasets when doing this.
                 volumes.append(
                     client.V1Volume(
                         name="dev-shm", empty_dir=client.V1EmptyDirVolumeSource(size_limit="32Gi", medium="Memory")

From a0ebd7d7e515797fc7dede96d7a04f486cefe614 Mon Sep 17 00:00:00 2001
From: github-actions <github-actions@github.com>
Date: Fri, 24 Apr 2026 10:46:32 +0000
Subject: [PATCH 05/29] Formatting and linting commit

---
 job_creator/jobcreator/job_creator.py | 19 ++++++++-----------
 1 file changed, 8 insertions(+), 11 deletions(-)

diff --git a/job_creator/jobcreator/job_creator.py b/job_creator/jobcreator/job_creator.py
index e76f680..425711d 100644
--- a/job_creator/jobcreator/job_creator.py
+++ b/job_creator/jobcreator/job_creator.py
@@ -6,7 +6,7 @@
 
 from kubernetes import client  # type: ignore[import-untyped]
 from kubernetes.client.exceptions import ApiException
-from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception
+from tenacity import retry, retry_if_exception, stop_after_attempt, wait_exponential
 
 from jobcreator.utils import load_kubernetes_config, logger
 
@@ -134,10 +134,10 @@ def _is_retryable_k8s_error(exc):
 
 
 @retry(
-        retry=retry_if_exception(_is_retryable_k8s_error),
-        stop=stop_after_attempt(3),
-        wait=wait_exponential(multiplier=1, min=1, max=10),
-        reraise=True,
+    retry=retry_if_exception(_is_retryable_k8s_error),
+    stop=stop_after_attempt(3),
+    wait=wait_exponential(multiplier=1, min=1, max=10),
+    reraise=True,
 )
 def _setup_ceph_pv(
     pv_name: str,
@@ -409,8 +409,8 @@ def spawn_job(  # noqa: PLR0913
                 )
                 volumes.append(client.V1Volume(name="imat-mount", persistent_volume_claim=imat_pvc_source))
                 volumes_mounts.append(client.V1VolumeMount(name="imat-mount", mount_path="/imat"))
-                # Because imat is special and uses mantid imaging to load large .tiff files, we need to 
-                # ensure the /dev/shm is larger than 64mb. We do however have a soft-ish limit of around 
+                # Because imat is special and uses mantid imaging to load large .tiff files, we need to
+                # ensure the /dev/shm is larger than 64mb. We do however have a soft-ish limit of around
                 # 32GiB on the size of datasets when doing this.
                 volumes.append(
                     client.V1Volume(
@@ -487,14 +487,11 @@ def spawn_job(  # noqa: PLR0913
             self._cleanup_resources(pv_names, pvc_names, job_namespace)
             raise
 
-
     def _cleanup_resources(self, pv_names, pvc_names, namespace):
         """Best-effort cleanup of created K8s resources."""
         for pvc_name in pvc_names:
             try:
-                client.CoreV1Api().delete_namespaced_persistent_valume_claim(
-                    name=pvc_name, namespace=namespace
-                )
+                client.CoreV1Api().delete_namespaced_persistent_valume_claim(name=pvc_name, namespace=namespace)
             except client.ApiException:
                 logger.warning("Failed to cleanup PVC: %s", pvc_name)
         for pv_name in pv_names:

From 5e0432ebf51b207a17e171a59d55ae3c0ef51261 Mon Sep 17 00:00:00 2001
From: eva-1729 <eva.maitland@stfc.ac.uk>
Date: Fri, 24 Apr 2026 12:02:36 +0100
Subject: [PATCH 06/29] Add types to functions

---
 job_creator/jobcreator/job_creator.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/job_creator/jobcreator/job_creator.py b/job_creator/jobcreator/job_creator.py
index e76f680..d71a5e0 100644
--- a/job_creator/jobcreator/job_creator.py
+++ b/job_creator/jobcreator/job_creator.py
@@ -5,7 +5,7 @@
 from typing import Any
 
 from kubernetes import client  # type: ignore[import-untyped]
-from kubernetes.client.exceptions import ApiException
+from kubernetes.client.exceptions import ApiException # type: ignore[import-untyped]
 from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception
 
 from jobcreator.utils import load_kubernetes_config, logger
@@ -125,7 +125,7 @@ def _setup_extras_pv(job_name: str, secret_namespace: str, manila_share_id: str,
     return pv_name
 
 
-def _is_retryable_k8s_error(exc):
+def _is_retryable_k8s_error(exc) -> bool:
     """Only retry on transient K8s API errors, not on conflicts or client errors."""
     if isinstance(exc, ApiException):
         return exc.status in (429, 500, 502, 503, 504)
@@ -488,7 +488,7 @@ def spawn_job(  # noqa: PLR0913
             raise
 
 
-    def _cleanup_resources(self, pv_names, pvc_names, namespace):
+    def _cleanup_resources(self, pv_names, pvc_names, namespace) -> None:
         """Best-effort cleanup of created K8s resources."""
         for pvc_name in pvc_names:
             try:

From 41dddca84247142c38fb05b5dbc816b1f5053791 Mon Sep 17 00:00:00 2001
From: github-actions <github-actions@github.com>
Date: Fri, 24 Apr 2026 11:04:28 +0000
Subject: [PATCH 07/29] Formatting and linting commit

---
 job_creator/jobcreator/job_creator.py | 5 ++---
 1 file changed, 2 insertions(+), 3 deletions(-)

diff --git a/job_creator/jobcreator/job_creator.py b/job_creator/jobcreator/job_creator.py
index 66b17e5..b8e4ec9 100644
--- a/job_creator/jobcreator/job_creator.py
+++ b/job_creator/jobcreator/job_creator.py
@@ -5,8 +5,8 @@
 from typing import Any
 
 from kubernetes import client  # type: ignore[import-untyped]
-from kubernetes.client.exceptions import ApiException # type: ignore[import-untyped]
-from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception
+from kubernetes.client.exceptions import ApiException  # type: ignore[import-untyped]
+from tenacity import retry, retry_if_exception, stop_after_attempt, wait_exponential
 
 from jobcreator.utils import load_kubernetes_config, logger
 
@@ -487,7 +487,6 @@ def spawn_job(  # noqa: PLR0913
             self._cleanup_resources(pv_names, pvc_names, job_namespace)
             raise
 
-
     def _cleanup_resources(self, pv_names, pvc_names, namespace) -> None:
         """Best-effort cleanup of created K8s resources."""
         for pvc_name in pvc_names:

From 0573baadc97e424d39b2edd9c9b76e6d9898bd87 Mon Sep 17 00:00:00 2001
From: eva-1729 <eva.maitland@stfc.ac.uk>
Date: Fri, 24 Apr 2026 12:54:55 +0100
Subject: [PATCH 08/29] Add type hints

---
 job_creator/jobcreator/job_creator.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/job_creator/jobcreator/job_creator.py b/job_creator/jobcreator/job_creator.py
index b8e4ec9..5c17faa 100644
--- a/job_creator/jobcreator/job_creator.py
+++ b/job_creator/jobcreator/job_creator.py
@@ -125,7 +125,7 @@ def _setup_extras_pv(job_name: str, secret_namespace: str, manila_share_id: str,
     return pv_name
 
 
-def _is_retryable_k8s_error(exc) -> bool:
+def _is_retryable_k8s_error(exc: ApiException) -> bool:
     """Only retry on transient K8s API errors, not on conflicts or client errors."""
     if isinstance(exc, ApiException):
         return exc.status in (429, 500, 502, 503, 504)
@@ -487,7 +487,7 @@ def spawn_job(  # noqa: PLR0913
             self._cleanup_resources(pv_names, pvc_names, job_namespace)
             raise
 
-    def _cleanup_resources(self, pv_names, pvc_names, namespace) -> None:
+    def _cleanup_resources(self, pv_names: list[str], pvc_names: list[str], namespace: str) -> None:
         """Best-effort cleanup of created K8s resources."""
         for pvc_name in pvc_names:
             try:

From e2ac6b76a8979af423c6109d08d29827d6c9733b Mon Sep 17 00:00:00 2001
From: eva-1729 <eva.maitland@stfc.ac.uk>
Date: Fri, 24 Apr 2026 14:15:12 +0100
Subject: [PATCH 09/29] Add retry for mkdir operations

---
 job_creator/jobcreator/utils.py | 12 ++++++++++++
 1 file changed, 12 insertions(+)

diff --git a/job_creator/jobcreator/utils.py b/job_creator/jobcreator/utils.py
index ed3109c..2bc1c31 100644
--- a/job_creator/jobcreator/utils.py
+++ b/job_creator/jobcreator/utils.py
@@ -59,6 +59,12 @@ def load_kubernetes_config() -> None:
             config.load_kube_config()
 
 
+@retry(
+    retry=retry_if_exception_type(OSError),
+    stop=stop_after_attempt(3),
+    wait=wait_exponential(multiplier=1, min=2, max=15),
+    reraise=True,
+)
 def create_ceph_mount_path_simple(
     user_number: str | None = None,
     experiment_number: str | None = None,
@@ -90,6 +96,12 @@ def create_ceph_mount_path_simple(
     return Path(mount_path) / ceph_path
 
 
+@retry(
+    retry=retry_if_exception_type(OSError),
+    stop=stop_after_attempt(3),
+    wait=wait_exponential(multiplier=1, min=2, max=15),
+    reraise=True,
+)
 def ensure_ceph_path_exists_autoreduction(ceph_path: Path) -> Path:
     """
     Takes a path that is intended to be on ceph and ensures that it will be correct for what we should mount and

From f4f8f6acc254dbbde972928f1b7ff15af95c9dab Mon Sep 17 00:00:00 2001
From: eva-1729 <eva.maitland@stfc.ac.uk>
Date: Fri, 24 Apr 2026 14:17:34 +0100
Subject: [PATCH 10/29] Add tenacity import

---
 job_creator/jobcreator/utils.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/job_creator/jobcreator/utils.py b/job_creator/jobcreator/utils.py
index 2bc1c31..4744ddc 100644
--- a/job_creator/jobcreator/utils.py
+++ b/job_creator/jobcreator/utils.py
@@ -11,6 +11,7 @@
 import requests
 from kubernetes import config  # type: ignore[import-untyped]
 from kubernetes.config import ConfigException  # type: ignore[import-untyped]
+from tenacity import retry, stop_after_attempt, wait_exponential
 
 stdout_handler = logging.StreamHandler(stream=sys.stdout)
 logging.basicConfig(

From d0913f620a5598e709dfcf4417cdfe5b8e4e5ee5 Mon Sep 17 00:00:00 2001
From: eva-1729 <eva.maitland@stfc.ac.uk>
Date: Fri, 24 Apr 2026 14:19:05 +0100
Subject: [PATCH 11/29] add import

---
 job_creator/jobcreator/utils.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/job_creator/jobcreator/utils.py b/job_creator/jobcreator/utils.py
index 4744ddc..16456bf 100644
--- a/job_creator/jobcreator/utils.py
+++ b/job_creator/jobcreator/utils.py
@@ -11,7 +11,7 @@
 import requests
 from kubernetes import config  # type: ignore[import-untyped]
 from kubernetes.config import ConfigException  # type: ignore[import-untyped]
-from tenacity import retry, stop_after_attempt, wait_exponential
+from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
 
 stdout_handler = logging.StreamHandler(stream=sys.stdout)
 logging.basicConfig(

From 867bc7764dbb2e190da098b34ea187fd7ab9c379 Mon Sep 17 00:00:00 2001
From: github-actions <github-actions@github.com>
Date: Fri, 24 Apr 2026 13:19:37 +0000
Subject: [PATCH 12/29] Formatting and linting commit

---
 job_creator/jobcreator/utils.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/job_creator/jobcreator/utils.py b/job_creator/jobcreator/utils.py
index 16456bf..59a9a83 100644
--- a/job_creator/jobcreator/utils.py
+++ b/job_creator/jobcreator/utils.py
@@ -11,7 +11,7 @@
 import requests
 from kubernetes import config  # type: ignore[import-untyped]
 from kubernetes.config import ConfigException  # type: ignore[import-untyped]
-from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
+from tenacity import retry, retry_if_exception_type, stop_after_attempt, wait_exponential
 
 stdout_handler = logging.StreamHandler(stream=sys.stdout)
 logging.basicConfig(

From 69df03dc8e15e7374d780250025184cdaf11496c Mon Sep 17 00:00:00 2001
From: eva-1729 <eva.maitland@stfc.ac.uk>
Date: Fri, 24 Apr 2026 14:23:07 +0100
Subject: [PATCH 13/29] Add retry to GHCR

---
 job_creator/jobcreator/utils.py | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/job_creator/jobcreator/utils.py b/job_creator/jobcreator/utils.py
index 59a9a83..884e2d1 100644
--- a/job_creator/jobcreator/utils.py
+++ b/job_creator/jobcreator/utils.py
@@ -157,6 +157,12 @@ def get_org_image_name_and_version_from_image_path(image_path: str) -> tuple[str
     return org_name, image_name, version  # Use organisation and image name without ghcr.io
 
 
+@retry(
+    retry=retry_if_exception_type((requests.ConnectionError, requests.Timeout)),
+    stop=stop_after_attempt(3),
+    wait=wait_exponential(multiplier=1, min=1, max=5),
+    reraise=True,
+)
 def get_sha256_using_image_from_ghcr(user_image: str, version: str = "") -> str:
     """
     Take the user image and request from the github api the sha256 of the image tag

From ddc761ea5701b2981f02df9cc3e5632cceca60cb Mon Sep 17 00:00:00 2001
From: eva-1729 <eva.maitland@stfc.ac.uk>
Date: Tue, 9 Jun 2026 11:09:29 +0100
Subject: [PATCH 14/29] Add some safety try and cleanup to queue_consumer in
 case of no connection

---
 job_creator/jobcreator/queue_consumer.py | 75 ++++++++++++++----------
 1 file changed, 45 insertions(+), 30 deletions(-)

diff --git a/job_creator/jobcreator/queue_consumer.py b/job_creator/jobcreator/queue_consumer.py
index c600839..9aec7dd 100644
--- a/job_creator/jobcreator/queue_consumer.py
+++ b/job_creator/jobcreator/queue_consumer.py
@@ -4,6 +4,7 @@
 
 import json
 import time
+import pika
 from collections.abc import Callable
 
 from pika import BlockingConnection, ConnectionParameters, PlainCredentials  # type: ignore[import-untyped]
@@ -34,24 +35,33 @@ def __init__(
         self.channel = None
         self.connect_to_broker()
 
-    def connect_to_broker(self) -> None:
+    def connect_to_broker(self, max_retries=10) -> None:
         """
         Use this to connect to the broker
         :return: None
         """
-        self.connection = BlockingConnection(self.connection_parameters)
-        self.channel = self.connection.channel()  # type: ignore[attr-defined]
-        self.channel.exchange_declare(  # type: ignore[attr-defined]
-            self.queue_name,
-            exchange_type="direct",
-            durable=True,
-        )
-        self.channel.queue_declare(  # type: ignore[attr-defined]
-            self.queue_name,
-            durable=True,
-            arguments={"x-queue-type": "quorum"},
-        )
-        self.channel.queue_bind(self.queue_name, self.queue_name, routing_key="")  # type: ignore[attr-defined]
+        for attempt in range(max_retries):
+            try:
+                self.connection = BlockingConnection(self.connection_parameters)
+                self.channel = self.connection.channel()  # type: ignore[attr-defined]
+                self.channel.exchange_declare(  # type: ignore[attr-defined]
+                    self.queue_name,
+                    exchange_type="direct",
+                    durable=True,
+                )
+                self.channel.queue_declare(  # type: ignore[attr-defined]
+                    self.queue_name,
+                    durable=True,
+                    arguments={"x-queue-type": "quorum"},
+                )
+                self.channel.queue_bind(self.queue_name, self.queue_name, routing_key="")  # type: ignore[attr-defined]
+            except pika.exceptions.AMQPConnectionError:
+                wait_time = min(2 ** attempt, 30)
+                logger.warning("Broker unavailable (attempt %d/%d), retrying in %ds",
+                               attempt + 1, max_retries, wait_time)
+                time.sleep(wait_time)
+        raise RuntimeError("Failed to connect to message broker after %d attempts" % max_retries)
+
 
     def _message_handler(self, msg: str) -> None:
         """
@@ -77,19 +87,24 @@ def start_consuming(self, callback_func: Callable[[], None], run_once: bool = Fa
         while run:
             if run_once:
                 run = False
-            callback_func()
-            for header, _, body in self.channel.consume(  # type: ignore[attr-defined]
-                self.queue_name,
-                inactivity_timeout=5,
-            ):
-                try:
-                    self._message_handler(body.decode())
-                    self.channel.basic_ack(header.delivery_tag)  # type: ignore[attr-defined]
-                except AttributeError:
-                    # If the message frame or body is missing attributes required e.g. the delivery tag
-                    pass
-                except Exception:
-                    logger.warning("Problem processing message: %s", body)
-                break
-
-            time.sleep(0.1)
+            try:
+                callback_func()
+                for header, _, body in self.channel.consume(  # type: ignore[attr-defined]
+                    self.queue_name,
+                    inactivity_timeout=5,
+                ):
+                    try:
+                        self._message_handler(body.decode())
+                        self.channel.basic_ack(header.delivery_tag)  # type: ignore[attr-defined]
+                    except AttributeError:
+                        # If the message frame or body is missing attributes required e.g. the delivery tag
+                        pass
+                    except Exception:
+                        logger.warning("Problem processing message: %s", body)
+                    break
+                time.sleep(0.1)
+            except (pika.exceptions.AMQPConnectionError,
+                    pika.exceptions.ChannelCloserByBroker) as e:
+                logger.warning("Lost connection to broker: %s. Reconnecting...",  e)
+                time.sleep(5)
+                self.connect_to_broker()

From dfaaaa8f60ea910eedab5ca9b4aa4e0a9422c959 Mon Sep 17 00:00:00 2001
From: eva-1729 <eva.maitland@stfc.ac.uk>
Date: Tue, 9 Jun 2026 12:46:03 +0100
Subject: [PATCH 15/29] ruff formatting fix

---
 job_creator/jobcreator/queue_consumer.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/job_creator/jobcreator/queue_consumer.py b/job_creator/jobcreator/queue_consumer.py
index 9aec7dd..d3e78cc 100644
--- a/job_creator/jobcreator/queue_consumer.py
+++ b/job_creator/jobcreator/queue_consumer.py
@@ -60,7 +60,7 @@ def connect_to_broker(self, max_retries=10) -> None:
                 logger.warning("Broker unavailable (attempt %d/%d), retrying in %ds",
                                attempt + 1, max_retries, wait_time)
                 time.sleep(wait_time)
-        raise RuntimeError("Failed to connect to message broker after %d attempts" % max_retries)
+        raise RuntimeError(f"Failed to connect to message broker after {max_retries} attempts")
 
 
     def _message_handler(self, msg: str) -> None:

From 85db9b8dc81ad344e4ebc16d88fc41532f8f387c Mon Sep 17 00:00:00 2001
From: github-actions <github-actions@github.com>
Date: Tue, 9 Jun 2026 11:46:35 +0000
Subject: [PATCH 16/29] Formatting and linting commit

---
 job_creator/jobcreator/queue_consumer.py | 15 +++++++--------
 1 file changed, 7 insertions(+), 8 deletions(-)

diff --git a/job_creator/jobcreator/queue_consumer.py b/job_creator/jobcreator/queue_consumer.py
index d3e78cc..fe8d43e 100644
--- a/job_creator/jobcreator/queue_consumer.py
+++ b/job_creator/jobcreator/queue_consumer.py
@@ -4,9 +4,9 @@
 
 import json
 import time
-import pika
 from collections.abc import Callable
 
+import pika
 from pika import BlockingConnection, ConnectionParameters, PlainCredentials  # type: ignore[import-untyped]
 
 from jobcreator.utils import logger
@@ -56,13 +56,13 @@ def connect_to_broker(self, max_retries=10) -> None:
                 )
                 self.channel.queue_bind(self.queue_name, self.queue_name, routing_key="")  # type: ignore[attr-defined]
             except pika.exceptions.AMQPConnectionError:
-                wait_time = min(2 ** attempt, 30)
-                logger.warning("Broker unavailable (attempt %d/%d), retrying in %ds",
-                               attempt + 1, max_retries, wait_time)
+                wait_time = min(2**attempt, 30)
+                logger.warning(
+                    "Broker unavailable (attempt %d/%d), retrying in %ds", attempt + 1, max_retries, wait_time
+                )
                 time.sleep(wait_time)
         raise RuntimeError(f"Failed to connect to message broker after {max_retries} attempts")
 
-
     def _message_handler(self, msg: str) -> None:
         """
         Handles a message from the message broker
@@ -103,8 +103,7 @@ def start_consuming(self, callback_func: Callable[[], None], run_once: bool = Fa
                         logger.warning("Problem processing message: %s", body)
                     break
                 time.sleep(0.1)
-            except (pika.exceptions.AMQPConnectionError,
-                    pika.exceptions.ChannelCloserByBroker) as e:
-                logger.warning("Lost connection to broker: %s. Reconnecting...",  e)
+            except (pika.exceptions.AMQPConnectionError, pika.exceptions.ChannelCloserByBroker) as e:
+                logger.warning("Lost connection to broker: %s. Reconnecting...", e)
                 time.sleep(5)
                 self.connect_to_broker()

From af904d02eae613756a9ef4a9c2af86d3b2061598 Mon Sep 17 00:00:00 2001
From: eva-1729 <eva.maitland@stfc.ac.uk>
Date: Tue, 9 Jun 2026 13:51:57 +0100
Subject: [PATCH 17/29] mypy fixes

---
 job_creator/jobcreator/queue_consumer.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/job_creator/jobcreator/queue_consumer.py b/job_creator/jobcreator/queue_consumer.py
index d3e78cc..cda7f82 100644
--- a/job_creator/jobcreator/queue_consumer.py
+++ b/job_creator/jobcreator/queue_consumer.py
@@ -7,7 +7,7 @@
 import pika
 from collections.abc import Callable
 
-from pika import BlockingConnection, ConnectionParameters, PlainCredentials  # type: ignore[import-untyped]
+from pika import BlockingConnection, ConnectionParameters, PlainCredentials
 
 from jobcreator.utils import logger
 
@@ -35,7 +35,7 @@ def __init__(
         self.channel = None
         self.connect_to_broker()
 
-    def connect_to_broker(self, max_retries=10) -> None:
+    def connect_to_broker(self, max_retries: int = 10) -> None:
         """
         Use this to connect to the broker
         :return: None

From 6e75af38dc2aa668007a8460f753d8e455110644 Mon Sep 17 00:00:00 2001
From: eva-1729 <eva.maitland@stfc.ac.uk>
Date: Tue, 9 Jun 2026 13:55:59 +0100
Subject: [PATCH 18/29] add pika import

---
 job_creator/jobcreator/queue_consumer.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/job_creator/jobcreator/queue_consumer.py b/job_creator/jobcreator/queue_consumer.py
index 437d5af..e8a63f0 100644
--- a/job_creator/jobcreator/queue_consumer.py
+++ b/job_creator/jobcreator/queue_consumer.py
@@ -6,6 +6,7 @@
 import time
 from collections.abc import Callable
 
+import pika
 from pika import BlockingConnection, ConnectionParameters, PlainCredentials
 
 from jobcreator.utils import logger

From 2192f29c180f9c453eb83724363961c2e5555fec Mon Sep 17 00:00:00 2001
From: eva-1729 <eva.maitland@stfc.ac.uk>
Date: Tue, 9 Jun 2026 14:02:07 +0100
Subject: [PATCH 19/29] add type ignore to pika again

---
 job_creator/jobcreator/queue_consumer.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/job_creator/jobcreator/queue_consumer.py b/job_creator/jobcreator/queue_consumer.py
index e8a63f0..3c25fba 100644
--- a/job_creator/jobcreator/queue_consumer.py
+++ b/job_creator/jobcreator/queue_consumer.py
@@ -6,7 +6,7 @@
 import time
 from collections.abc import Callable
 
-import pika
+import pika #type: ignore[import]
 from pika import BlockingConnection, ConnectionParameters, PlainCredentials
 
 from jobcreator.utils import logger

From faac08c760f59c3207b3d0135cc517fdc513424e Mon Sep 17 00:00:00 2001
From: github-actions <github-actions@github.com>
Date: Tue, 9 Jun 2026 13:02:40 +0000
Subject: [PATCH 20/29] Formatting and linting commit

---
 job_creator/jobcreator/queue_consumer.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/job_creator/jobcreator/queue_consumer.py b/job_creator/jobcreator/queue_consumer.py
index 3c25fba..429c964 100644
--- a/job_creator/jobcreator/queue_consumer.py
+++ b/job_creator/jobcreator/queue_consumer.py
@@ -6,7 +6,7 @@
 import time
 from collections.abc import Callable
 
-import pika #type: ignore[import]
+import pika  # type: ignore[import]
 from pika import BlockingConnection, ConnectionParameters, PlainCredentials
 
 from jobcreator.utils import logger

From d7a76d00387dd377fea584df812610f63d129b25 Mon Sep 17 00:00:00 2001
From: eva-1729 <eva.maitland@stfc.ac.uk>
Date: Tue, 9 Jun 2026 14:04:02 +0100
Subject: [PATCH 21/29] fix type ignore

---
 job_creator/jobcreator/queue_consumer.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/job_creator/jobcreator/queue_consumer.py b/job_creator/jobcreator/queue_consumer.py
index 3c25fba..0a2fa38 100644
--- a/job_creator/jobcreator/queue_consumer.py
+++ b/job_creator/jobcreator/queue_consumer.py
@@ -6,7 +6,7 @@
 import time
 from collections.abc import Callable
 
-import pika #type: ignore[import]
+import pika #type: ignore[import-untyped]
 from pika import BlockingConnection, ConnectionParameters, PlainCredentials
 
 from jobcreator.utils import logger

From 486cb0567610dafd70d6f1fdab8f9f8ea33cc76f Mon Sep 17 00:00:00 2001
From: github-actions <github-actions@github.com>
Date: Tue, 9 Jun 2026 13:05:07 +0000
Subject: [PATCH 22/29] Formatting and linting commit

---
 job_creator/jobcreator/queue_consumer.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/job_creator/jobcreator/queue_consumer.py b/job_creator/jobcreator/queue_consumer.py
index 0a2fa38..2249993 100644
--- a/job_creator/jobcreator/queue_consumer.py
+++ b/job_creator/jobcreator/queue_consumer.py
@@ -6,7 +6,7 @@
 import time
 from collections.abc import Callable
 
-import pika #type: ignore[import-untyped]
+import pika  # type: ignore[import-untyped]
 from pika import BlockingConnection, ConnectionParameters, PlainCredentials
 
 from jobcreator.utils import logger

From 46823f147e0cf28f7ad157b8ee1946a9a6572aca Mon Sep 17 00:00:00 2001
From: eva-1729 <eva.maitland@stfc.ac.uk>
Date: Wed, 10 Jun 2026 14:37:42 +0100
Subject: [PATCH 23/29] Fix test failures

---
 job_creator/jobcreator/queue_consumer.py | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/job_creator/jobcreator/queue_consumer.py b/job_creator/jobcreator/queue_consumer.py
index 2249993..d60c9b5 100644
--- a/job_creator/jobcreator/queue_consumer.py
+++ b/job_creator/jobcreator/queue_consumer.py
@@ -55,6 +55,8 @@ def connect_to_broker(self, max_retries: int = 10) -> None:
                     arguments={"x-queue-type": "quorum"},
                 )
                 self.channel.queue_bind(self.queue_name, self.queue_name, routing_key="")  # type: ignore[attr-defined]
+                logger.info("Connected to broker")
+                return
             except pika.exceptions.AMQPConnectionError:
                 wait_time = min(2**attempt, 30)
                 logger.warning(
@@ -103,7 +105,7 @@ def start_consuming(self, callback_func: Callable[[], None], run_once: bool = Fa
                         logger.warning("Problem processing message: %s", body)
                     break
                 time.sleep(0.1)
-            except (pika.exceptions.AMQPConnectionError, pika.exceptions.ChannelCloserByBroker) as e:
+            except (pika.exceptions.AMQPConnectionError, pika.exceptions.ChannelClosedByBroker) as e:
                 logger.warning("Lost connection to broker: %s. Reconnecting...", e)
                 time.sleep(5)
                 self.connect_to_broker()

From 5ae0722ed6521037ab2a9850c04f1df8d8f8f7de Mon Sep 17 00:00:00 2001
From: eva-1729 <eva.maitland@stfc.ac.uk>
Date: Mon, 22 Jun 2026 15:46:00 +0100
Subject: [PATCH 24/29] Add more test coverage with antigravity Claude Opus 4.6

---
 job_creator/jobcreator/job_creator.py   |   2 +-
 job_creator/test/test_job_creator.py    | 153 ++++++++++++++++++++++++
 job_creator/test/test_queue_consumer.py | 119 ++++++++++++++++++
 3 files changed, 273 insertions(+), 1 deletion(-)

diff --git a/job_creator/jobcreator/job_creator.py b/job_creator/jobcreator/job_creator.py
index 5c17faa..0eefc4d 100644
--- a/job_creator/jobcreator/job_creator.py
+++ b/job_creator/jobcreator/job_creator.py
@@ -491,7 +491,7 @@ def _cleanup_resources(self, pv_names: list[str], pvc_names: list[str], namespac
         """Best-effort cleanup of created K8s resources."""
         for pvc_name in pvc_names:
             try:
-                client.CoreV1Api().delete_namespaced_persistent_valume_claim(name=pvc_name, namespace=namespace)
+                client.CoreV1Api().delete_namespaced_persistent_volume_claim(name=pvc_name, namespace=namespace)
             except client.ApiException:
                 logger.warning("Failed to cleanup PVC: %s", pvc_name)
         for pv_name in pv_names:
diff --git a/job_creator/test/test_job_creator.py b/job_creator/test/test_job_creator.py
index 32dc99f..3a21b35 100644
--- a/job_creator/test/test_job_creator.py
+++ b/job_creator/test/test_job_creator.py
@@ -2,8 +2,12 @@
 from unittest import mock
 from unittest.mock import call
 
+import pytest
+from kubernetes.client.exceptions import ApiException
+
 from jobcreator.job_creator import (
     JobCreator,
+    _is_retryable_k8s_error,
     _setup_ceph_pv,
     _setup_extras_pv,
     _setup_extras_pvc,
@@ -665,3 +669,152 @@ def test_jobcreator_spawn_job_dev_mode_false(
     assert client.V1PersistentVolumeClaimVolumeSource.call_count == 2  # noqa: PLR2004
 
     setup_ceph_pv.assert_not_called()
+
+
+# --- Tests for _is_retryable_k8s_error ---
+
+
+@pytest.mark.parametrize("status_code", [429, 500, 502, 503, 504])
+def test_is_retryable_k8s_error_returns_true_for_transient_status_codes(status_code):
+    exc = ApiException(status=status_code)
+    assert _is_retryable_k8s_error(exc) is True
+
+
+@pytest.mark.parametrize("status_code", [400, 401, 403, 404, 409, 422])
+def test_is_retryable_k8s_error_returns_false_for_non_transient_status_codes(status_code):
+    exc = ApiException(status=status_code)
+    assert _is_retryable_k8s_error(exc) is False
+
+
+def test_is_retryable_k8s_error_returns_true_for_connection_error():
+    assert _is_retryable_k8s_error(ConnectionError("connection refused")) is True
+
+
+def test_is_retryable_k8s_error_returns_true_for_timeout_error():
+    assert _is_retryable_k8s_error(TimeoutError("timed out")) is True
+
+
+def test_is_retryable_k8s_error_returns_false_for_unrelated_exception():
+    assert _is_retryable_k8s_error(ValueError("something else")) is False
+
+
+def test_is_retryable_k8s_error_returns_false_for_runtime_error():
+    assert _is_retryable_k8s_error(RuntimeError("unexpected")) is False
+
+
+# --- Tests for _cleanup_resources ---
+
+
+@mock.patch("jobcreator.job_creator.client")
+@mock.patch("jobcreator.job_creator.load_kubernetes_config")
+def test_cleanup_resources_deletes_pvcs_and_pvs(_, client):
+    job_creator = JobCreator("sha", False)
+    pv_names = ["pv-1", "pv-2"]
+    pvc_names = ["pvc-1", "pvc-2"]
+    namespace = "test-ns"
+
+    job_creator._cleanup_resources(pv_names, pvc_names, namespace)
+
+    core_api = client.CoreV1Api.return_value
+    assert core_api.delete_namespaced_persistent_volume_claim.call_count == 2  # noqa: PLR2004
+    core_api.delete_namespaced_persistent_volume_claim.assert_any_call(name="pvc-1", namespace="test-ns")
+    core_api.delete_namespaced_persistent_volume_claim.assert_any_call(name="pvc-2", namespace="test-ns")
+    assert core_api.delete_persistent_volume.call_count == 2  # noqa: PLR2004
+    core_api.delete_persistent_volume.assert_any_call(name="pv-1")
+    core_api.delete_persistent_volume.assert_any_call(name="pv-2")
+
+
+@mock.patch("jobcreator.job_creator.logger")
+@mock.patch("jobcreator.job_creator.client")
+@mock.patch("jobcreator.job_creator.load_kubernetes_config")
+def test_cleanup_resources_logs_warning_on_pvc_deletion_failure(_, client, mock_logger):
+    client.ApiException = ApiException
+    core_api = client.CoreV1Api.return_value
+    core_api.delete_namespaced_persistent_volume_claim.side_effect = ApiException(status=404)
+
+    job_creator = JobCreator("sha", False)
+    job_creator._cleanup_resources(["pv-1"], ["pvc-1"], "test-ns")
+
+    mock_logger.warning.assert_any_call("Failed to cleanup PVC: %s", "pvc-1")
+    # PV cleanup should still proceed
+    core_api.delete_persistent_volume.assert_called_once_with(name="pv-1")
+
+
+@mock.patch("jobcreator.job_creator.logger")
+@mock.patch("jobcreator.job_creator.client")
+@mock.patch("jobcreator.job_creator.load_kubernetes_config")
+def test_cleanup_resources_logs_warning_on_pv_deletion_failure(_, client, mock_logger):
+    client.ApiException = ApiException
+    core_api = client.CoreV1Api.return_value
+    core_api.delete_persistent_volume.side_effect = ApiException(status=404)
+
+    job_creator = JobCreator("sha", False)
+    job_creator._cleanup_resources(["pv-1"], [], "test-ns")
+
+    mock_logger.warning.assert_any_call("Failed to cleanup PV: %s", "pv-1")
+
+
+@mock.patch("jobcreator.job_creator.client")
+@mock.patch("jobcreator.job_creator.load_kubernetes_config")
+def test_cleanup_resources_handles_empty_lists(_, client):
+    job_creator = JobCreator("sha", False)
+    job_creator._cleanup_resources([], [], "test-ns")
+
+    core_api = client.CoreV1Api.return_value
+    core_api.delete_namespaced_persistent_volume_claim.assert_not_called()
+    core_api.delete_persistent_volume.assert_not_called()
+
+
+# --- Tests for spawn_job cleanup-on-failure ---
+
+
+@mock.patch("jobcreator.job_creator._setup_extras_pv")
+@mock.patch("jobcreator.job_creator._setup_extras_pvc")
+@mock.patch("jobcreator.job_creator._setup_smb_pv")
+@mock.patch("jobcreator.job_creator._setup_pvc")
+@mock.patch("jobcreator.job_creator._setup_ceph_pv")
+@mock.patch("jobcreator.job_creator.load_kubernetes_config")
+@mock.patch("jobcreator.job_creator.client")
+def test_spawn_job_calls_cleanup_and_reraises_on_failure(
+    client,
+    _,  # noqa: PT019
+    setup_ceph_pv,
+    setup_pvc,
+    setup_smb_pv,
+    setup_extras_pvc,
+    setup_extras_pv,
+):
+    """When job creation fails, cleanup_resources is called and the exception is re-raised."""
+    client.BatchV1Api.return_value.create_namespaced_job.side_effect = ApiException(status=500)
+    job_creator = JobCreator("sha", False)
+    job_creator._cleanup_resources = mock.MagicMock()
+    job_name = "test-job"
+
+    with pytest.raises(ApiException):
+        job_creator.spawn_job(
+            job_name,
+            "script",
+            "namespace",
+            "ceph-secret",
+            "ceph-ns",
+            "cluster-id",
+            "fs-name",
+            "/ceph/path",
+            1,
+            3600,
+            "fia-api-host",
+            "api-key",
+            "runner-image",
+            "manila-share-id",
+            "manila-access-id",
+            [],
+            [],
+            None,
+        )
+
+    job_creator._cleanup_resources.assert_called_once()
+    call_args = job_creator._cleanup_resources.call_args
+    # Verify pv_names and pvc_names lists were passed (non-empty since setup steps succeeded)
+    assert len(call_args[0][0]) > 0  # pv_names
+    assert len(call_args[0][1]) > 0  # pvc_names
+    assert call_args[0][2] == "namespace"  # namespace
diff --git a/job_creator/test/test_queue_consumer.py b/job_creator/test/test_queue_consumer.py
index fe772f9..36e4430 100644
--- a/job_creator/test/test_queue_consumer.py
+++ b/job_creator/test/test_queue_consumer.py
@@ -107,3 +107,122 @@ def raise_exception():
         quc.start_consuming(mock.MagicMock(), run_once=True)
 
     logger.warning.assert_called_once_with("Problem processing message: %s", body)
+
+
+# --- Tests for connect_to_broker retry logic ---
+
+
+@mock.patch("jobcreator.queue_consumer.time")
+@mock.patch("jobcreator.queue_consumer.ConnectionParameters")
+@mock.patch("jobcreator.queue_consumer.BlockingConnection")
+def test_connect_to_broker_retries_on_amqp_error_then_succeeds(blocking_connection, _conn_params, mock_time):
+    """connect_to_broker retries on AMQPConnectionError and succeeds on a subsequent attempt."""
+    import pika.exceptions
+
+    blocking_connection.side_effect = [
+        pika.exceptions.AMQPConnectionError("refused"),
+        pika.exceptions.AMQPConnectionError("refused"),
+        mock.MagicMock(),  # Third attempt succeeds
+    ]
+
+    quc = QueueConsumer(mock.MagicMock(), mock.MagicMock(), mock.MagicMock(), mock.MagicMock(), mock.MagicMock())
+
+    assert quc.connection is not None
+    assert quc.channel is not None
+    assert blocking_connection.call_count == 3  # noqa: PLR2004
+    # Verify sleep was called for the two failed attempts
+    assert mock_time.sleep.call_count >= 2  # noqa: PLR2004
+
+
+@mock.patch("jobcreator.queue_consumer.time")
+@mock.patch("jobcreator.queue_consumer.ConnectionParameters")
+@mock.patch("jobcreator.queue_consumer.BlockingConnection")
+def test_connect_to_broker_raises_runtime_error_after_max_retries(blocking_connection, _conn_params, mock_time):
+    """connect_to_broker raises RuntimeError after exhausting all retry attempts."""
+    import pika.exceptions
+
+    blocking_connection.side_effect = pika.exceptions.AMQPConnectionError("refused")
+
+    with pytest.raises(RuntimeError, match="Failed to connect to message broker after 10 attempts"):
+        QueueConsumer(mock.MagicMock(), mock.MagicMock(), mock.MagicMock(), mock.MagicMock(), mock.MagicMock())
+
+
+@mock.patch("jobcreator.queue_consumer.time")
+@mock.patch("jobcreator.queue_consumer.ConnectionParameters")
+@mock.patch("jobcreator.queue_consumer.BlockingConnection")
+def test_connect_to_broker_uses_exponential_backoff(blocking_connection, _conn_params, mock_time):
+    """connect_to_broker sleeps with exponentially increasing wait times."""
+    import pika.exceptions
+
+    blocking_connection.side_effect = pika.exceptions.AMQPConnectionError("refused")
+
+    with pytest.raises(RuntimeError):
+        QueueConsumer(mock.MagicMock(), mock.MagicMock(), mock.MagicMock(), mock.MagicMock(), mock.MagicMock())
+
+    # Verify exponential backoff: min(2^0, 30)=1, min(2^1, 30)=2, min(2^2, 30)=4, ...
+    sleep_calls = [c[0][0] for c in mock_time.sleep.call_args_list]
+    for i, wait in enumerate(sleep_calls):
+        assert wait == min(2**i, 30)
+
+
+# --- Tests for start_consuming reconnection logic ---
+
+
+@mock.patch("jobcreator.queue_consumer.time")
+@mock.patch("jobcreator.queue_consumer.ConnectionParameters")
+@mock.patch("jobcreator.queue_consumer.BlockingConnection")
+def test_start_consuming_reconnects_on_amqp_connection_error(blocking_connection, _conn_params, mock_time):
+    """start_consuming reconnects to the broker when AMQPConnectionError is raised during consuming."""
+    import pika.exceptions
+
+    quc = QueueConsumer(mock.MagicMock(), mock.MagicMock(), mock.MagicMock(), mock.MagicMock(), mock.MagicMock())
+    quc.connect_to_broker = mock.MagicMock()
+
+    call_count = 0
+
+    def callback_side_effect():
+        nonlocal call_count
+        call_count += 1
+        if call_count == 1:
+            raise pika.exceptions.AMQPConnectionError("lost connection")
+
+    callback = mock.MagicMock(side_effect=callback_side_effect)
+
+    # run_once=True won't help here since we need the while loop to iterate.
+    # Instead, have the second callback call succeed and use channel.consume normally.
+    # The reconnect path sets run=False on second iteration via run_once logic.
+    # Actually, let's just test that connect_to_broker is called after the error.
+    # We need to break the loop, so after reconnection, raise KeyboardInterrupt.
+    quc.connect_to_broker.side_effect = KeyboardInterrupt
+
+    with pytest.raises(KeyboardInterrupt):
+        quc.start_consuming(callback, run_once=False)
+
+    quc.connect_to_broker.assert_called_once()
+
+
+@mock.patch("jobcreator.queue_consumer.time")
+@mock.patch("jobcreator.queue_consumer.ConnectionParameters")
+@mock.patch("jobcreator.queue_consumer.BlockingConnection")
+def test_start_consuming_reconnects_on_channel_closed_by_broker(blocking_connection, _conn_params, mock_time):
+    """start_consuming reconnects when ChannelClosedByBroker is raised."""
+    import pika.exceptions
+
+    quc = QueueConsumer(mock.MagicMock(), mock.MagicMock(), mock.MagicMock(), mock.MagicMock(), mock.MagicMock())
+    quc.connect_to_broker = mock.MagicMock()
+
+    call_count = 0
+
+    def callback_side_effect():
+        nonlocal call_count
+        call_count += 1
+        if call_count == 1:
+            raise pika.exceptions.ChannelClosedByBroker(406, "PRECONDITION_FAILED")
+
+    callback = mock.MagicMock(side_effect=callback_side_effect)
+    quc.connect_to_broker.side_effect = KeyboardInterrupt
+
+    with pytest.raises(KeyboardInterrupt):
+        quc.start_consuming(callback, run_once=False)
+
+    quc.connect_to_broker.assert_called_once()

From 13d57c71a8d6bc77bdaacb7ffe6e322856e4c8cb Mon Sep 17 00:00:00 2001
From: eva-1729 <eva.maitland@stfc.ac.uk>
Date: Mon, 22 Jun 2026 16:08:24 +0100
Subject: [PATCH 25/29] linting fixes

---
 job_creator/test/test_job_creator.py    |  8 ++++----
 job_creator/test/test_queue_consumer.py | 17 +++++++----------
 2 files changed, 11 insertions(+), 14 deletions(-)

diff --git a/job_creator/test/test_job_creator.py b/job_creator/test/test_job_creator.py
index 3a21b35..31a43bd 100644
--- a/job_creator/test/test_job_creator.py
+++ b/job_creator/test/test_job_creator.py
@@ -707,7 +707,7 @@ def test_is_retryable_k8s_error_returns_false_for_runtime_error():
 
 @mock.patch("jobcreator.job_creator.client")
 @mock.patch("jobcreator.job_creator.load_kubernetes_config")
-def test_cleanup_resources_deletes_pvcs_and_pvs(_, client):
+def test_cleanup_resources_deletes_pvcs_and_pvs(mock_load_k8s_config, client):
     job_creator = JobCreator("sha", False)
     pv_names = ["pv-1", "pv-2"]
     pvc_names = ["pvc-1", "pvc-2"]
@@ -727,7 +727,7 @@ def test_cleanup_resources_deletes_pvcs_and_pvs(_, client):
 @mock.patch("jobcreator.job_creator.logger")
 @mock.patch("jobcreator.job_creator.client")
 @mock.patch("jobcreator.job_creator.load_kubernetes_config")
-def test_cleanup_resources_logs_warning_on_pvc_deletion_failure(_, client, mock_logger):
+def test_cleanup_resources_logs_warning_on_pvc_deletion_failure(mock_load_k8s_config, client, mock_logger):
     client.ApiException = ApiException
     core_api = client.CoreV1Api.return_value
     core_api.delete_namespaced_persistent_volume_claim.side_effect = ApiException(status=404)
@@ -743,7 +743,7 @@ def test_cleanup_resources_logs_warning_on_pvc_deletion_failure(_, client, mock_
 @mock.patch("jobcreator.job_creator.logger")
 @mock.patch("jobcreator.job_creator.client")
 @mock.patch("jobcreator.job_creator.load_kubernetes_config")
-def test_cleanup_resources_logs_warning_on_pv_deletion_failure(_, client, mock_logger):
+def test_cleanup_resources_logs_warning_on_pv_deletion_failure(mock_load_k8s_config, client, mock_logger):
     client.ApiException = ApiException
     core_api = client.CoreV1Api.return_value
     core_api.delete_persistent_volume.side_effect = ApiException(status=404)
@@ -756,7 +756,7 @@ def test_cleanup_resources_logs_warning_on_pv_deletion_failure(_, client, mock_l
 
 @mock.patch("jobcreator.job_creator.client")
 @mock.patch("jobcreator.job_creator.load_kubernetes_config")
-def test_cleanup_resources_handles_empty_lists(_, client):
+def test_cleanup_resources_handles_empty_lists(mock_load_k8s_config, client):
     job_creator = JobCreator("sha", False)
     job_creator._cleanup_resources([], [], "test-ns")
 
diff --git a/job_creator/test/test_queue_consumer.py b/job_creator/test/test_queue_consumer.py
index 36e4430..6c532ca 100644
--- a/job_creator/test/test_queue_consumer.py
+++ b/job_creator/test/test_queue_consumer.py
@@ -1,5 +1,6 @@
 from unittest import mock
 
+import pika.exceptions
 import pytest
 from pika import PlainCredentials
 
@@ -115,9 +116,9 @@ def raise_exception():
 @mock.patch("jobcreator.queue_consumer.time")
 @mock.patch("jobcreator.queue_consumer.ConnectionParameters")
 @mock.patch("jobcreator.queue_consumer.BlockingConnection")
-def test_connect_to_broker_retries_on_amqp_error_then_succeeds(blocking_connection, _conn_params, mock_time):
+def test_connect_to_broker_retries_on_amqp_error_then_succeeds(blocking_connection, mock_conn_params, mock_time):
     """connect_to_broker retries on AMQPConnectionError and succeeds on a subsequent attempt."""
-    import pika.exceptions
+
 
     blocking_connection.side_effect = [
         pika.exceptions.AMQPConnectionError("refused"),
@@ -137,9 +138,8 @@ def test_connect_to_broker_retries_on_amqp_error_then_succeeds(blocking_connecti
 @mock.patch("jobcreator.queue_consumer.time")
 @mock.patch("jobcreator.queue_consumer.ConnectionParameters")
 @mock.patch("jobcreator.queue_consumer.BlockingConnection")
-def test_connect_to_broker_raises_runtime_error_after_max_retries(blocking_connection, _conn_params, mock_time):
+def test_connect_to_broker_raises_runtime_error_after_max_retries(blocking_connection, mock_conn_params, mock_time):
     """connect_to_broker raises RuntimeError after exhausting all retry attempts."""
-    import pika.exceptions
 
     blocking_connection.side_effect = pika.exceptions.AMQPConnectionError("refused")
 
@@ -150,9 +150,8 @@ def test_connect_to_broker_raises_runtime_error_after_max_retries(blocking_conne
 @mock.patch("jobcreator.queue_consumer.time")
 @mock.patch("jobcreator.queue_consumer.ConnectionParameters")
 @mock.patch("jobcreator.queue_consumer.BlockingConnection")
-def test_connect_to_broker_uses_exponential_backoff(blocking_connection, _conn_params, mock_time):
+def test_connect_to_broker_uses_exponential_backoff(blocking_connection, mock_conn_params, mock_time):
     """connect_to_broker sleeps with exponentially increasing wait times."""
-    import pika.exceptions
 
     blocking_connection.side_effect = pika.exceptions.AMQPConnectionError("refused")
 
@@ -171,9 +170,8 @@ def test_connect_to_broker_uses_exponential_backoff(blocking_connection, _conn_p
 @mock.patch("jobcreator.queue_consumer.time")
 @mock.patch("jobcreator.queue_consumer.ConnectionParameters")
 @mock.patch("jobcreator.queue_consumer.BlockingConnection")
-def test_start_consuming_reconnects_on_amqp_connection_error(blocking_connection, _conn_params, mock_time):
+def test_start_consuming_reconnects_on_amqp_connection_error(blocking_connection, mock_conn_params, mock_time):
     """start_consuming reconnects to the broker when AMQPConnectionError is raised during consuming."""
-    import pika.exceptions
 
     quc = QueueConsumer(mock.MagicMock(), mock.MagicMock(), mock.MagicMock(), mock.MagicMock(), mock.MagicMock())
     quc.connect_to_broker = mock.MagicMock()
@@ -204,9 +202,8 @@ def callback_side_effect():
 @mock.patch("jobcreator.queue_consumer.time")
 @mock.patch("jobcreator.queue_consumer.ConnectionParameters")
 @mock.patch("jobcreator.queue_consumer.BlockingConnection")
-def test_start_consuming_reconnects_on_channel_closed_by_broker(blocking_connection, _conn_params, mock_time):
+def test_start_consuming_reconnects_on_channel_closed_by_broker(blocking_connection, mock_conn_params, mock_time):
     """start_consuming reconnects when ChannelClosedByBroker is raised."""
-    import pika.exceptions
 
     quc = QueueConsumer(mock.MagicMock(), mock.MagicMock(), mock.MagicMock(), mock.MagicMock(), mock.MagicMock())
     quc.connect_to_broker = mock.MagicMock()

From 33b722c71dbdf7a6b8d144fafe3bb678f3236518 Mon Sep 17 00:00:00 2001
From: github-actions <github-actions@github.com>
Date: Mon, 22 Jun 2026 15:08:59 +0000
Subject: [PATCH 26/29] Formatting and linting commit

---
 job_creator/test/test_queue_consumer.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/job_creator/test/test_queue_consumer.py b/job_creator/test/test_queue_consumer.py
index 6c532ca..1a714ac 100644
--- a/job_creator/test/test_queue_consumer.py
+++ b/job_creator/test/test_queue_consumer.py
@@ -119,7 +119,6 @@ def raise_exception():
 def test_connect_to_broker_retries_on_amqp_error_then_succeeds(blocking_connection, mock_conn_params, mock_time):
     """connect_to_broker retries on AMQPConnectionError and succeeds on a subsequent attempt."""
 
-
     blocking_connection.side_effect = [
         pika.exceptions.AMQPConnectionError("refused"),
         pika.exceptions.AMQPConnectionError("refused"),

From 9f8ace7874f5bd3dfa1f6c0d5d2fcccb9ec1a9d1 Mon Sep 17 00:00:00 2001
From: eva-1729 <eva.maitland@stfc.ac.uk>
Date: Mon, 22 Jun 2026 16:18:35 +0100
Subject: [PATCH 27/29] change scope of fixture

---
 job_creator/test/test_queue_consumer.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/job_creator/test/test_queue_consumer.py b/job_creator/test/test_queue_consumer.py
index 6c532ca..6ecfcca 100644
--- a/job_creator/test/test_queue_consumer.py
+++ b/job_creator/test/test_queue_consumer.py
@@ -13,7 +13,7 @@
 QUEUE_NAME = mock.MagicMock()
 
 
-@pytest.fixture(autouse=True, scope="module")
+@pytest.fixture(autouse=True)
 def setup_queue_consumer():
     with (
         mock.patch("jobcreator.queue_consumer.ConnectionParameters") as connection_parameters,

From a3258b93079f7d9e31d5162b0190ca6fa56ff33f Mon Sep 17 00:00:00 2001
From: eva-1729 <eva.maitland@stfc.ac.uk>
Date: Mon, 22 Jun 2026 16:32:35 +0100
Subject: [PATCH 28/29] formatting

---
 job_creator/jobcreator/job_creator.py | 50 +++++++++++++--------------
 1 file changed, 25 insertions(+), 25 deletions(-)

diff --git a/job_creator/jobcreator/job_creator.py b/job_creator/jobcreator/job_creator.py
index 45c577e..4848530 100644
--- a/job_creator/jobcreator/job_creator.py
+++ b/job_creator/jobcreator/job_creator.py
@@ -428,23 +428,23 @@ def spawn_job(  # noqa: PLR0913
             )
             volumes_mounts.append(client.V1VolumeMount(name="dev-shm", mount_path="/dev/shm"))  # noqa: S108
 
-        # Decide whether this is a GPU workload. IMAT jobs run mantid imaging on GPU nodes and need
-        # the NVIDIA runtime + a GPU resource request so the GPU Operator injects the matching
-        # userspace driver libraries (libcuda.so.*) into the container.
-        gpu_job = "imat" in special_pvs
-
-        main_container = client.V1Container(
-            name=job_name,
-            image=runner_image,
-            args=[script],
-            env=[client.V1EnvVar(name="PYTHONUNBUFFERED", value="1")],
-            volume_mounts=volumes_mounts,
-            resources=client.V1ResourceRequirements(
-                limits={"nvidia.com/gpu": "1"},
+            # Decide whether this is a GPU workload. IMAT jobs run mantid imaging on GPU nodes and need
+            # the NVIDIA runtime + a GPU resource request so the GPU Operator injects the matching
+            # userspace driver libraries (libcuda.so.*) into the container.
+            gpu_job = "imat" in special_pvs
+
+            main_container = client.V1Container(
+                name=job_name,
+                image=runner_image,
+                args=[script],
+                env=[client.V1EnvVar(name="PYTHONUNBUFFERED", value="1")],
+                volume_mounts=volumes_mounts,
+                resources=client.V1ResourceRequirements(
+                    limits={"nvidia.com/gpu": "1"},
+                )
+                if gpu_job
+                else None,
             )
-            if gpu_job
-            else None,
-        )
 
             watcher_container = client.V1Container(
                 name="job-watcher",
@@ -462,15 +462,15 @@ def spawn_job(  # noqa: PLR0913
             affinity = _generate_affinities(node_affinity_dict=affinity)
             tolerations = _generate_tolerations_from_taints(taints)
 
-        pod_spec = client.V1PodSpec(
-            affinity=affinity,
-            service_account_name="jobwatcher",
-            containers=[main_container, watcher_container],
-            restart_policy="Never",
-            tolerations=tolerations,
-            volumes=volumes,
-            runtime_class_name="nvidia" if gpu_job else None,
-        )
+            pod_spec = client.V1PodSpec(
+                affinity=affinity,
+                service_account_name="jobwatcher",
+                containers=[main_container, watcher_container],
+                restart_policy="Never",
+                tolerations=tolerations,
+                volumes=volumes,
+                runtime_class_name="nvidia" if gpu_job else None,
+            )
 
             pod_metadata = client.V1ObjectMeta(
                 labels={"reduce.isis.cclrc.ac.uk/job-source": "automated-reduction"},

From 521d766d9e5313871371875fc017fb01f0381fa0 Mon Sep 17 00:00:00 2001
From: eva-1729 <eva.maitland@stfc.ac.uk>
Date: Mon, 22 Jun 2026 16:36:56 +0100
Subject: [PATCH 29/29] remove duplicate code block

---
 job_creator/jobcreator/job_creator.py | 9 ---------
 1 file changed, 9 deletions(-)

diff --git a/job_creator/jobcreator/job_creator.py b/job_creator/jobcreator/job_creator.py
index 4848530..0ec4700 100644
--- a/job_creator/jobcreator/job_creator.py
+++ b/job_creator/jobcreator/job_creator.py
@@ -419,15 +419,6 @@ def spawn_job(  # noqa: PLR0913
                 )
                 volumes_mounts.append(client.V1VolumeMount(name="dev-shm", mount_path="/dev/shm"))  # noqa: S108
 
-            main_container = client.V1Container(
-                name=job_name,
-                image=runner_image,
-                args=[script],
-                env=[client.V1EnvVar(name="PYTHONUNBUFFERED", value="1")],
-                volume_mounts=volumes_mounts,
-            )
-            volumes_mounts.append(client.V1VolumeMount(name="dev-shm", mount_path="/dev/shm"))  # noqa: S108
-
             # Decide whether this is a GPU workload. IMAT jobs run mantid imaging on GPU nodes and need
             # the NVIDIA runtime + a GPU resource request so the GPU Operator injects the matching
             # userspace driver libraries (libcuda.so.*) into the container.