diff --git a/ami/base/models.py b/ami/base/models.py
index 8254a70d1..fd1ebc68f 100644
--- a/ami/base/models.py
+++ b/ami/base/models.py
@@ -1,4 +1,6 @@
+from django.contrib.auth.models import AbstractUser, AnonymousUser
 from django.db import models
+from guardian.shortcuts import get_perms
 
 import ami.tasks
 
@@ -29,5 +31,94 @@ def update_calculated_fields(self, *args, **kwargs):
         """Update calculated fields specific to each model."""
         pass
 
+    def _get_object_perms(self, user):
+        """
+        Get the object-level permissions for the user on this instance.
+        This method retrieves permissions like `update_modelname`, `create_modelname`, etc.
+        """
+        project = self.get_project()
+        if not project:
+            return []
+
+        model_name = self._meta.model_name
+        all_perms = get_perms(user, project)
+        object_perms = [perm for perm in all_perms if perm.endswith(f"_{model_name}")]
+        return object_perms
+
+    def check_permission(self, user: AbstractUser | AnonymousUser, action: str) -> bool:
+        """
+        Check if the user has permission to perform the action
+        on this instance.
+        This method is used to determine if the user can perform
+        CRUD operations or custom actions on the model instance.
+        """
+        project = self.get_project() if hasattr(self, "get_project") else None
+        if not project:
+            return False
+        if action == "retrieve":
+            # Allow view
+            return True
+
+        model = self._meta.model_name
+        crud_map = {
+            "create": f"create_{model}",
+            "update": f"update_{model}",
+            "partial_update": f"update_{model}",
+            "destroy": f"delete_{model}",
+        }
+
+        if action in crud_map:
+            return user.has_perm(crud_map[action], project)
+
+        # Delegate to model-specific logic
+        return self.check_custom_permission(user, action)
+
+    def check_custom_permission(self, user: AbstractUser | AnonymousUser, action: str) -> bool:
+        """Check custom permissions for the user on this instance.
+        This is used for actions that are not standard CRUD operations.
+        """
+        assert self._meta.model_name is not None, "Model must have a model_name defined in Meta class."
+        model_name = self._meta.model_name.lower()
+        permission_codename = f"{action}_{model_name}"
+        project = self.get_project() if hasattr(self, "get_project") else None
+
+        return user.has_perm(permission_codename, project)
+
+    def get_user_object_permissions(self, user) -> list[str]:
+        """
+        Returns a list of object-level permissions the user has on this instance.
+        This is used by frontend to determine what actions the user can perform.
+        """
+        # Return all permissions for superusers
+        if user.is_superuser:
+            allowed_custom_actions = self.get_custom_user_permissions(user)
+            return ["update", "delete"] + allowed_custom_actions
+
+        object_perms = self._get_object_perms(user)
+        # Check for update and delete permissions
+        allowed_actions = set()
+        for perm in object_perms:
+            action = perm.split("_", 1)[0]
+            if action in {"update", "delete"}:
+                allowed_actions.add(action)
+
+        allowed_custom_actions = self.get_custom_user_permissions(user)
+        allowed_actions.update(set(allowed_custom_actions))
+        return list(allowed_actions)
+
+    def get_custom_user_permissions(self, user: AbstractUser | AnonymousUser) -> list[str]:
+        """
+        Returns a list of custom permissions (not standard CRUD actions) that the user has on this instance.
+        """
+        object_perms = self._get_object_perms(user)
+        custom_perms = set()
+        # Extract custom permissions that are not standard CRUD actions
+        for perm in object_perms:
+            action = perm.split("_", 1)[0]
+            # Make sure to exclude standard CRUD actions
+            if action not in ["view", "create", "update", "delete"]:
+                custom_perms.add(action)
+        return list(custom_perms)
+
     class Meta:
         abstract = True
diff --git a/ami/base/permissions.py b/ami/base/permissions.py
index 249df0ac2..723b1e58e 100644
--- a/ami/base/permissions.py
+++ b/ami/base/permissions.py
@@ -6,19 +6,7 @@
 from guardian.shortcuts import get_perms
 from rest_framework import permissions
 
-from ami.jobs.models import Job
-from ami.main.models import (
-    BaseModel,
-    Deployment,
-    Device,
-    Project,
-    S3StorageSource,
-    Site,
-    SourceImage,
-    SourceImageCollection,
-    SourceImageUpload,
-)
-from ami.users.roles import ProjectManager
+from ami.main.models import BaseModel
 
 logger = logging.getLogger(__name__)
 
@@ -64,18 +52,8 @@ def add_object_level_permissions(
     """
 
     permissions = response_data.get("user_permissions", set())
-    project = instance.get_project() if hasattr(instance, "get_project") else None
-    model_name = instance._meta.model_name  # Get model name
-    if user and user.is_superuser:
-        permissions.update(["update", "delete"])
-
-    if project:
-        user_permissions = get_perms(user, project)
-        # Filter and extract only the action part of "action_modelname" based on instance type
-        filtered_permissions = filter_permissions(permissions=user_permissions, model_name=model_name)
-        # Do not return create, view permissions at object-level
-        filtered_permissions -= {"create", "view"}
-        permissions.update(filtered_permissions)
+    if isinstance(instance, BaseModel):
+        permissions.update(instance.get_user_object_permissions(user))
     response_data["user_permissions"] = list(permissions)
     return response_data
 
@@ -99,165 +77,13 @@ def add_collection_level_permissions(user: User | None, response_data: dict, mod
     return response_data
 
 
-class CRUDPermission(permissions.BasePermission):
+class ObjectPermission(permissions.BasePermission):
     """
-    Generic CRUD permission class that dynamically checks user permissions on an object.
-    Permission names follow the convention: `create_<model>`, `update_<model>`, `delete_<model>`.
+    Generic permission class that delegates to the model's `check_permission(user, action)` method.
     """
 
-    model = None
-
     def has_permission(self, request, view):
-        """Handles general permission checks"""
-
-        return True  # Fallback to object level permissions
-
-    def has_object_permission(self, request, view, obj):
-        """Handles object-level permission checks."""
-        model_name = self.model._meta.model_name
-        project = obj.get_project() if hasattr(obj, "get_project") else None
-        # check for create action
-        if view.action == "create":
-            return request.user.is_superuser or request.user.has_perm(f"create_{model_name}", project)
-
-        if view.action == "retrieve":
-            return True  # Allow all users to view objects
-
-        # Map ViewSet actions to permission names
-        action_perms = {
-            "retrieve": f"view_{model_name}",
-            "update": f"update_{model_name}",
-            "partial_update": f"update_{model_name}",
-            "destroy": f"delete_{model_name}",
-        }
-
-        required_perm = action_perms.get(view.action)
-        if not required_perm:
-            return True
-        return request.user.has_perm(required_perm, project)
-
-
-class ProjectCRUDPermission(CRUDPermission):
-    model = Project
-
-
-class JobCRUDPermission(CRUDPermission):
-    model = Job
-
-
-class DeploymentCRUDPermission(CRUDPermission):
-    model = Deployment
-
-
-class SourceImageCollectionCRUDPermission(CRUDPermission):
-    model = SourceImageCollection
-
-
-class SourceImageUploadCRUDPermission(CRUDPermission):
-    model = SourceImageUpload
-
-
-class SourceImageCRUDPermission(CRUDPermission):
-    model = SourceImage
-
-
-class CanStarSourceImage(permissions.BasePermission):
-    """Custom permission to check if the user can star a Source image."""
-
-    permission = Project.Permissions.STAR_SOURCE_IMAGE
-
-    def has_object_permission(self, request, view, obj):
-        if view.action in ["unstar", "star"]:
-            project = obj.get_project() if hasattr(obj, "get_project") else None
-            return request.user.has_perm(self.permission, project)
-        return True
-
-
-class S3StorageSourceCRUDPermission(CRUDPermission):
-    model = S3StorageSource
-
-
-class SiteCRUDPermission(CRUDPermission):
-    model = Site
-
-
-class DeviceCRUDPermission(CRUDPermission):
-    model = Device
-
-
-# Identification permission checks
-class CanUpdateIdentification(permissions.BasePermission):
-    """Custom permission to check if the user can update/create an identification."""
-
-    permission = Project.Permissions.UPDATE_IDENTIFICATION
-
-    def has_object_permission(self, request, view, obj):
-        if view.action in ["create", "update", "partial_update"]:
-            project = obj.get_project() if hasattr(obj, "get_project") else None
-            return request.user.has_perm(self.permission, project)
-        return True
-
-
-class CanDeleteIdentification(permissions.BasePermission):
-    """Custom permission to check if the user can delete an identification."""
-
-    permission = Project.Permissions.DELETE_IDENTIFICATION
-
-    def has_object_permission(self, request, view, obj):
-        project = obj.get_project() if hasattr(obj, "get_project") else None
-        # Check if user is superuser or staff or project manager
-        if view.action == "destroy":
-            if request.user.is_superuser or ProjectManager.has_role(request.user, project):
-                return True
-            # Check if the user is the owner of the object
-            return obj.user == request.user
-        return True
-
-
-# Job run permission check
-class CanRunJob(permissions.BasePermission):
-    """Custom permission to check if the user can run a job."""
-
-    permission = Project.Permissions.RUN_JOB
-
-    def has_object_permission(self, request, view, obj):
-        if view.action == "run":
-            project = obj.get_project() if hasattr(obj, "get_project") else None
-            return request.user.has_perm(self.permission, project)
-        return True
-
-
-class CanRetryJob(permissions.BasePermission):
-    """Custom permission to check if the user can retry a job."""
-
-    permission = Project.Permissions.RETRY_JOB
-
-    def has_object_permission(self, request, view, obj):
-        if view.action == "retry":
-            project = obj.get_project() if hasattr(obj, "get_project") else None
-            return request.user.has_perm(self.permission, project)
-        return True
-
-
-class CanCancelJob(permissions.BasePermission):
-    """Custom permission to check if the user can cancel a job."""
-
-    permission = Project.Permissions.CANCEL_JOB
-
-    def has_object_permission(self, request, view, obj):
-        if view.action == "cancel":
-            project = obj.get_project() if hasattr(obj, "get_project") else None
-            return request.user.has_perm(self.permission, project)
-        return True
-
-
-class CanPopulateSourceImageCollection(permissions.BasePermission):
-    """Custom permission to check if the user can populate a collection."""
-
-    permission = Project.Permissions.POPULATE_COLLECTION
+        return True  # Always allow — object-level handles actual checks
 
-    def has_object_permission(self, request, view, obj):
-        if view.action == "populate":
-            project = obj.get_project() if hasattr(obj, "get_project") else None
-            return request.user.has_perm(self.permission, project)
-        return True
+    def has_object_permission(self, request, view, obj: BaseModel):
+        return obj.check_permission(request.user, view.action)
diff --git a/ami/jobs/models.py b/ami/jobs/models.py
index 1166f2c74..aa3215068 100644
--- a/ami/jobs/models.py
+++ b/ami/jobs/models.py
@@ -11,6 +11,7 @@
 from django.db import models, transaction
 from django.utils.text import slugify
 from django_pydantic_field import SchemaField
+from guardian.shortcuts import get_perms
 
 from ami.base.models import BaseModel
 from ami.base.schemas import ConfigurableStage, ConfigurableStageParam
@@ -907,6 +908,37 @@ def save(self, update_progress=True, *args, **kwargs):
         if self.progress.summary.status != self.status:
             logger.warning(f"Job {self} status mismatches progress: {self.progress.summary.status} != {self.status}")
 
+    def check_custom_permission(self, user, action: str) -> bool:
+        job_type = self.job_type_key.lower()
+        if self.source_image_single:
+            action = "run_single_image"
+        if action in ["run", "cancel", "retry"]:
+            permission_codename = f"run_{job_type}_job"
+        else:
+            permission_codename = f"{action}_{job_type}_job"
+
+        project = self.get_project() if hasattr(self, "get_project") else None
+        return user.has_perm(permission_codename, project)
+
+    def get_custom_user_permissions(self, user) -> list[str]:
+        project = self.get_project()
+        if not project:
+            return []
+
+        custom_perms = set()
+        model_name = "job"
+        perms = get_perms(user, project)
+        job_type = self.job_type_key.lower()
+        for perm in perms:
+            # permissions are in the format "action_modelname"
+            if perm.endswith(f"{job_type}_{model_name}"):
+                action = perm[: -len(f"_{job_type}_{model_name}")]
+                # make sure to exclude standard CRUD actions
+                if action not in ["view", "create", "update", "delete"]:
+                    custom_perms.add(action)
+        logger.debug(f"Custom permissions for user {user} on project {self}, with jobtype {job_type}: {custom_perms}")
+        return list(custom_perms)
+
     @classmethod
     def default_progress(cls) -> JobProgress:
         """Return the progress of each stage of this job as a dictionary"""
diff --git a/ami/jobs/tests.py b/ami/jobs/tests.py
index 6623141f3..64fbf23a2 100644
--- a/ami/jobs/tests.py
+++ b/ami/jobs/tests.py
@@ -155,7 +155,7 @@ def test_run_job(self):
         self.client.force_authenticate(user=self.user)
         # give user run permission
 
-        assign_perm(Project.Permissions.RUN_JOB, self.user, self.project)
+        assign_perm(Project.Permissions.RUN_POPULATE_CAPTURES_COLLECTION_JOB, self.user, self.project)
 
         resp = self.client.post(jobs_run_url)
         self.assertEqual(resp.status_code, 200)
diff --git a/ami/jobs/views.py b/ami/jobs/views.py
index 48f288dac..5fffdb6fd 100644
--- a/ami/jobs/views.py
+++ b/ami/jobs/views.py
@@ -5,9 +5,10 @@
 from django.utils import timezone
 from drf_spectacular.utils import extend_schema
 from rest_framework.decorators import action
+from rest_framework.exceptions import PermissionDenied
 from rest_framework.response import Response
 
-from ami.base.permissions import CanCancelJob, CanRetryJob, CanRunJob, JobCRUDPermission
+from ami.base.permissions import ObjectPermission
 from ami.base.views import ProjectMixin
 from ami.main.api.views import DefaultViewSet
 from ami.utils.fields import url_boolean_param
@@ -66,7 +67,8 @@ class JobViewSet(DefaultViewSet, ProjectMixin):
         "source_image_collection",
         "pipeline",
     ]
-    permission_classes = [CanRunJob, CanRetryJob, CanCancelJob, JobCRUDPermission]
+
+    permission_classes = [ObjectPermission]
 
     def get_serializer_class(self):
         """
@@ -129,8 +131,12 @@ def perform_create(self, serializer):
 
         job: Job = serializer.save()  # type: ignore
         if url_boolean_param(self.request, "start_now", default=False):
-            # job.run()
-            job.enqueue()
+            if job.check_custom_permission(self.request.user, "run"):
+                # If the user has permission, enqueue the job
+                job.enqueue()
+            else:
+                # If the user does not have permission, raise an error
+                raise PermissionDenied("You do not have permission to run this job.")
 
     def get_queryset(self) -> QuerySet:
         jobs = super().get_queryset()
diff --git a/ami/main/api/views.py b/ami/main/api/views.py
index fb95c90dd..acdfdf5cb 100644
--- a/ami/main/api/views.py
+++ b/ami/main/api/views.py
@@ -25,21 +25,7 @@
 
 from ami.base.filters import NullsLastOrderingFilter, ThresholdFilter
 from ami.base.pagination import LimitOffsetPaginationWithPermissions
-from ami.base.permissions import (
-    CanDeleteIdentification,
-    CanPopulateSourceImageCollection,
-    CanStarSourceImage,
-    CanUpdateIdentification,
-    DeploymentCRUDPermission,
-    DeviceCRUDPermission,
-    IsActiveStaffOrReadOnly,
-    ProjectCRUDPermission,
-    S3StorageSourceCRUDPermission,
-    SiteCRUDPermission,
-    SourceImageCollectionCRUDPermission,
-    SourceImageCRUDPermission,
-    SourceImageUploadCRUDPermission,
-)
+from ami.base.permissions import IsActiveStaffOrReadOnly, ObjectPermission
 from ami.base.serializers import FilterParamsSerializer, SingleParamSerializer
 from ami.base.views import ProjectMixin
 from ami.main.api.serializers import TagSerializer
@@ -151,7 +137,7 @@ class ProjectViewSet(DefaultViewSet, ProjectMixin):
     queryset = Project.objects.filter(active=True).prefetch_related("deployments").all()
     serializer_class = ProjectSerializer
     pagination_class = ProjectPagination
-    permission_classes = [ProjectCRUDPermission]
+    permission_classes = [ObjectPermission]
 
     def get_queryset(self):
         qs: ProjectQuerySet = super().get_queryset()  # type: ignore
@@ -218,7 +204,7 @@ class DeploymentViewSet(DefaultViewSet, ProjectMixin):
         "last_date",
     ]
 
-    permission_classes = [DeploymentCRUDPermission]
+    permission_classes = [ObjectPermission]
 
     def get_serializer_class(self):
         """
@@ -481,7 +467,7 @@ class SourceImageViewSet(DefaultViewSet, ProjectMixin):
         "deployment__name",
         "event__start",
     ]
-    permission_classes = [CanStarSourceImage, SourceImageCRUDPermission]
+    permission_classes = [ObjectPermission]
 
     def get_serializer_class(self):
         """
@@ -641,8 +627,7 @@ class SourceImageCollectionViewSet(DefaultViewSet, ProjectMixin):
     )
     serializer_class = SourceImageCollectionSerializer
     permission_classes = [
-        CanPopulateSourceImageCollection,
-        SourceImageCollectionCRUDPermission,
+        ObjectPermission,
     ]
     filterset_fields = ["method"]
     ordering_fields = [
@@ -759,7 +744,7 @@ class SourceImageUploadViewSet(DefaultViewSet, ProjectMixin):
     queryset = SourceImageUpload.objects.all()
 
     serializer_class = SourceImageUploadSerializer
-    permission_classes = [SourceImageUploadCRUDPermission]
+    permission_classes = [ObjectPermission]
     require_project = True
 
     def get_queryset(self) -> QuerySet:
@@ -1666,7 +1651,8 @@ class IdentificationViewSet(DefaultViewSet):
         "updated_at",
         "user",
     ]
-    permission_classes = [CanUpdateIdentification, CanDeleteIdentification]
+
+    permission_classes = [ObjectPermission]
 
     def perform_create(self, serializer):
         """
@@ -1694,7 +1680,7 @@ class SiteViewSet(DefaultViewSet, ProjectMixin):
         "updated_at",
         "name",
     ]
-    permission_classes = [SiteCRUDPermission]
+    permission_classes = [ObjectPermission]
 
     def get_queryset(self) -> QuerySet:
         query_set: QuerySet = super().get_queryset()
@@ -1721,7 +1707,7 @@ class DeviceViewSet(DefaultViewSet, ProjectMixin):
         "updated_at",
         "name",
     ]
-    permission_classes = [DeviceCRUDPermission]
+    permission_classes = [ObjectPermission]
 
     def get_queryset(self) -> QuerySet:
         query_set: QuerySet = super().get_queryset()
@@ -1753,7 +1739,7 @@ class StorageSourceViewSet(DefaultViewSet, ProjectMixin):
         "updated_at",
         "name",
     ]
-    permission_classes = [S3StorageSourceCRUDPermission]
+    permission_classes = [ObjectPermission]
 
     def get_queryset(self) -> QuerySet:
         query_set: QuerySet = super().get_queryset()
diff --git a/ami/main/migrations/0060_alter_sourceimagecollection_method.py b/ami/main/migrations/0060_alter_sourceimagecollection_method.py
index a32610c63..1f59c3875 100644
--- a/ami/main/migrations/0060_alter_sourceimagecollection_method.py
+++ b/ami/main/migrations/0060_alter_sourceimagecollection_method.py
@@ -5,7 +5,7 @@
 
 class Migration(migrations.Migration):
     dependencies = [
-        ("main", "0059_alter_project_options"),
+        ("main", "0058_alter_project_options"),
     ]
 
     operations = [
diff --git a/ami/main/migrations/0067_delete_deprecated_permissions.py b/ami/main/migrations/0067_delete_deprecated_permissions.py
new file mode 100644
index 000000000..a300ea666
--- /dev/null
+++ b/ami/main/migrations/0067_delete_deprecated_permissions.py
@@ -0,0 +1,34 @@
+from django.db import migrations
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+def delete_deprecated_permissions(apps, schema_editor):
+    Permission = apps.get_model("auth", "Permission")
+    ContentType = apps.get_model("contenttypes", "ContentType")
+
+    deprecated_codenames = [
+        "process_sourceimage",
+        "process_single_image_job",
+        "process_single_image_ml_job",
+        "run_job",
+        "retry_job",
+        "cancel_job",
+    ]
+
+    permissions = Permission.objects.filter(codename__in=deprecated_codenames)
+    for perm in permissions:
+        logger.info(f"Deleting permission: {perm.codename}")
+        perm.delete()
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("main", "0066_alter_project_feature_flags_and_more"),
+    ]
+
+    operations = [
+        migrations.RunPython(delete_deprecated_permissions, reverse_code=migrations.RunPython.noop),
+    ]
diff --git a/ami/main/migrations/0059_alter_project_options.py b/ami/main/migrations/0068_alter_project_options.py
similarity index 82%
rename from ami/main/migrations/0059_alter_project_options.py
rename to ami/main/migrations/0068_alter_project_options.py
index 8cfd9c6f9..1c69b8c79 100644
--- a/ami/main/migrations/0059_alter_project_options.py
+++ b/ami/main/migrations/0068_alter_project_options.py
@@ -1,11 +1,11 @@
-# Generated by Django 4.2.10 on 2025-04-02 10:53
+# Generated by Django 4.2.10 on 2025-08-14 21:11
 
 from django.db import migrations
 
 
 class Migration(migrations.Migration):
     dependencies = [
-        ("main", "0058_alter_project_options"),
+        ("main", "0067_delete_deprecated_permissions"),
     ]
 
     operations = [
@@ -19,10 +19,12 @@ class Migration(migrations.Migration):
                     ("delete_identification", "Can delete identifications"),
                     ("create_job", "Can create a job"),
                     ("update_job", "Can update a job"),
-                    ("run_job", "Can run a job"),
+                    ("run_ml_job", "Can run/retry/cancel ML jobs"),
+                    ("run_populate_captures_collection_job", "Can run/retry/cancel Populate Collection jobs"),
+                    ("run_data_storage_sync_job", "Can run/retry/cancel Data Storage Sync jobs"),
+                    ("run_data_export_job", "Can run/retry/cancel Data Export jobs"),
+                    ("run_single_image_ml_job", "Can process a single capture"),
                     ("delete_job", "Can delete a job"),
-                    ("retry_job", "Can retry a job"),
-                    ("cancel_job", "Can cancel a job"),
                     ("create_deployment", "Can create a deployment"),
                     ("delete_deployment", "Can delete a deployment"),
                     ("update_deployment", "Can update a deployment"),
diff --git a/ami/main/migrations/0069_merge_20250818_1201.py b/ami/main/migrations/0069_merge_20250818_1201.py
new file mode 100644
index 000000000..4d3d7d9fc
--- /dev/null
+++ b/ami/main/migrations/0069_merge_20250818_1201.py
@@ -0,0 +1,12 @@
+# Generated by Django 4.2.10 on 2025-08-18 12:01
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("main", "0067_alter_project_feature_flags"),
+        ("main", "0068_alter_project_options"),
+    ]
+
+    operations = []
diff --git a/ami/main/models.py b/ami/main/models.py
index 52c3c2a46..c4864e709 100644
--- a/ami/main/models.py
+++ b/ami/main/models.py
@@ -13,6 +13,7 @@
 import pydantic
 from django.apps import apps
 from django.conf import settings
+from django.contrib.auth.models import AbstractUser, AnonymousUser
 from django.contrib.postgres.fields import ArrayField
 from django.core.exceptions import ValidationError
 from django.core.files.storage import default_storage
@@ -24,6 +25,7 @@
 from django.template.defaultfilters import filesizeformat
 from django.utils import timezone
 from django_pydantic_field import SchemaField
+from guardian.shortcuts import get_perms
 
 import ami.tasks
 import ami.utils
@@ -304,10 +306,12 @@ class Permissions:
         # Job permissions
         CREATE_JOB = "create_job"
         UPDATE_JOB = "update_job"
-        RUN_JOB = "run_job"
+        RUN_ML_JOB = "run_ml_job"
+        RUN_SINGLE_IMAGE_JOB = "run_single_image_ml_job"
+        RUN_POPULATE_CAPTURES_COLLECTION_JOB = "run_populate_captures_collection_job"
+        RUN_DATA_STORAGE_SYNC_JOB = "run_data_storage_sync_job"
+        RUN_DATA_EXPORT_JOB = "run_data_export_job"
         DELETE_JOB = "delete_job"
-        RETRY_JOB = "retry_job"
-        CANCEL_JOB = "cancel_job"
 
         # Deployment permissions
         CREATE_DEPLOYMENT = "create_deployment"
@@ -362,10 +366,12 @@ class Meta:
             # Job permissions
             ("create_job", "Can create a job"),
             ("update_job", "Can update a job"),
-            ("run_job", "Can run a job"),
+            ("run_ml_job", "Can run/retry/cancel ML jobs"),
+            ("run_populate_captures_collection_job", "Can run/retry/cancel Populate Collection jobs"),
+            ("run_data_storage_sync_job", "Can run/retry/cancel Data Storage Sync jobs"),
+            ("run_data_export_job", "Can run/retry/cancel Data Export jobs"),
+            ("run_single_image_ml_job", "Can process a single capture"),
             ("delete_job", "Can delete a job"),
-            ("retry_job", "Can retry a job"),
-            ("cancel_job", "Can cancel a job"),
             # Deployment permissions
             ("create_deployment", "Can create a deployment"),
             ("delete_deployment", "Can delete a deployment"),
@@ -1726,6 +1732,30 @@ def save(self, update_calculated_fields=True, *args, **kwargs):
         if update_calculated_fields:
             self.update_calculated_fields(save=True)
 
+    def check_custom_permission(self, user, action: str) -> bool:
+        project = self.get_project() if hasattr(self, "get_project") else None
+        if action in ["star", "unstar"]:
+            return user.has_perm(Project.Permissions.STAR_SOURCE_IMAGE, project)
+
+    def get_custom_user_permissions(self, user) -> list[str]:
+        project = self.get_project()
+        if not project:
+            return []
+
+        custom_perms = set()
+        perms = get_perms(user, project)
+        for perm in perms:
+            # permissions are in the format "action_modelname"
+            if perm.endswith("_sourceimage"):
+                # process_single_image_sourceimage
+                action = perm.split("_", 1)[0]
+                # make sure to exclude standard CRUD actions
+                if action not in ["view", "create", "update", "delete"]:
+                    custom_perms.add(action)
+        if Project.Permissions.RUN_SINGLE_IMAGE_JOB in perms:
+            custom_perms.add(Project.Permissions.RUN_SINGLE_IMAGE_JOB)
+        return list(custom_perms)
+
     class Meta:
         ordering = ("deployment", "event", "timestamp")
 
@@ -2035,6 +2065,21 @@ def delete(self, *args, **kwargs):
         # Allow the update_occurrence_determination to determine the next best ID
         update_occurrence_determination(self.occurrence, current_determination=self.taxon)
 
+    def check_permission(self, user: AbstractUser | AnonymousUser, action: str) -> bool:
+        """Custom permission check logic for Identification model."""
+        import ami.users.roles as roles
+
+        project = self.get_project()
+        if not project:
+            return False
+
+        if action == "destroy":
+            # Allow if user is superuser, project manager, or owner of the identification
+            return user.is_superuser or roles.ProjectManager.has_role(user, project) or self.user == user
+
+        # Fallback to base class permission checks
+        return super().check_permission(user, action)
+
 
 @final
 class ClassificationResult(BaseModel):
diff --git a/ami/main/tests.py b/ami/main/tests.py
index 445960ba6..51329b232 100644
--- a/ami/main/tests.py
+++ b/ami/main/tests.py
@@ -5,13 +5,13 @@
 from django.core.files.uploadedfile import SimpleUploadedFile
 from django.db import connection, models
 from django.test import TestCase, override_settings
-from guardian.shortcuts import get_perms
+from guardian.shortcuts import assign_perm, get_perms, remove_perm
 from PIL import Image
 from rest_framework import status
 from rest_framework.test import APIRequestFactory, APITestCase
 from rich import print
 
-from ami.jobs.models import Job
+from ami.jobs.models import VALID_JOB_TYPES, Job
 from ami.main.models import (
     Deployment,
     Device,
@@ -1288,7 +1288,15 @@ def setUp(self) -> None:
                 "sourceimageupload": {"create": True, "update": True, "delete": True},
                 "site": {"create": True, "update": True, "delete": True},
                 "device": {"create": True, "update": True, "delete": True},
-                "job": {"create": True, "update": True, "delete": True, "run": True, "retry": True, "cancel": True},
+                "job": {
+                    "create": True,
+                    "update": True,
+                    "delete": True,
+                    "run_single_image": True,
+                    "run": False,
+                    "retry": False,
+                    "cancel": False,
+                },
                 "identification": {"create": True, "update": True, "delete": True},
                 "capture": {"star": True, "unstar": True},
             },
@@ -1301,9 +1309,10 @@ def setUp(self) -> None:
                 "sourceimageupload": {"create": False, "update": False, "delete": False},
                 "device": {"create": False, "update": False, "delete": False},
                 "job": {
-                    "create": False,
+                    "create": True,
                     "update": False,
                     "delete": False,
+                    "run_single_image": True,
                     "run": False,
                     "retry": False,
                     "cancel": False,
@@ -1320,9 +1329,10 @@ def setUp(self) -> None:
                 "site": {"create": False, "update": False, "delete": False},
                 "device": {"create": False, "update": False, "delete": False},
                 "job": {
-                    "create": False,
+                    "create": True,
                     "update": False,
                     "delete": False,
+                    "run_single_image": True,
                     "run": False,
                     "retry": False,
                     "cancel": False,
@@ -1342,6 +1352,7 @@ def setUp(self) -> None:
                     "create": False,
                     "update": False,
                     "delete": False,
+                    "run_single_image": False,
                     "run": False,
                     "retry": False,
                     "cancel": False,
@@ -1369,7 +1380,7 @@ def _create_project(self, owner):
         )
 
     def _create_job(self):
-        self.job = Job.objects.create(name="Test Job", project=self.project)
+        self.job = Job.objects.create(name="Test Job", project=self.project, job_type_key="ml")
 
     def _create_source_image_upload_file(self):
         image_buffer = BytesIO()
@@ -1499,7 +1510,11 @@ def _test_role_permissions(self, role_class, user, permissions_map):
                 object_id = entity_ids[entity]
                 obj = next((r for r in results if r["id"] == object_id), None)
                 if obj:
-                    self.assertEqual(set(obj.get("user_permissions", [])), set(object_permissions))
+                    self.assertEqual(
+                        set(obj.get("user_permissions", [])),
+                        set(object_permissions),
+                        f"Object permissions mismatch for {entity}",
+                    )
 
             # Step 2: Test Update
             logger.info(f"Testing {role_class} update permission for {entity} , actions {actions}")
@@ -1523,7 +1538,11 @@ def _test_role_permissions(self, role_class, user, permissions_map):
                     if action in actions:
                         response = self.client.post(f"{endpoints[entity]}{entity_ids[entity]}/{action}/")
                         expected_status = status.HTTP_200_OK if actions[action] else status.HTTP_403_FORBIDDEN
-                        self.assertEqual(response.status_code, expected_status)
+                        self.assertEqual(
+                            response.status_code,
+                            expected_status,
+                            f"{role_class} {action} permission failed for {entity}",
+                        )
 
             if entity == "collection" and entity_ids[entity] and "populate" in actions:
                 logger.info(f"Testing {role_class} for  collection populate custom permission")
@@ -1536,12 +1555,12 @@ def _test_role_permissions(self, role_class, user, permissions_map):
             can_star = permissions_map["capture"].get("star", False)
             response = self.client.post(endpoints["capture_star"])
             expected_status = status.HTTP_200_OK if can_star else status.HTTP_403_FORBIDDEN
-            self.assertEqual(response.status_code, expected_status)
+            self.assertEqual(response.status_code, expected_status, f"{role_class} star permission failed")
             logger.info(f"Testing {role_class} for  capture unstar permission ")
             can_unstar = permissions_map["capture"].get("unstar", False)
             response = self.client.post(endpoints["capture_unstar"])
             expected_status = status.HTTP_200_OK if can_unstar else status.HTTP_403_FORBIDDEN
-            self.assertEqual(response.status_code, expected_status)
+            self.assertEqual(response.status_code, expected_status, f"{role_class} unstar permission failed")
         logger.info(f"{role_class}: entity_ids: {entity_ids}")
         # Step 5: Unassign Role and Verify Permissions are Revoked
         if role_class:
@@ -1595,7 +1614,7 @@ def _test_role_permissions(self, role_class, user, permissions_map):
                     response = self.client.delete(f"{endpoints[entity]}{entity_ids[entity]}/")
                     logger.info(f"{role_class} delete response status for {entity} : {response.status_code}")
                     expected_status = status.HTTP_204_NO_CONTENT if can_delete else status.HTTP_403_FORBIDDEN
-                    self.assertEqual(response.status_code, expected_status)
+                    self.assertEqual(response.status_code, expected_status, f"Delete permission failed for {entity}")
 
             # try to delete the project
             entity = "project"
@@ -1785,3 +1804,157 @@ def test_sync_creates_events_and_updates_counts(self):
             "Deployment events_count should reflect updated event count",
         )
         logger.info(f"Initial events count: {initial_events_count}, Updated events count: {updated_events.count()}")
+
+
+class TestFineGrainedJobRunPermission(APITestCase):
+    def setUp(self):
+        super().setUp()
+        self.user = User.objects.create_user(
+            email="regularuser@insectai.org",
+            password="password123",
+        )
+        self.client.force_authenticate(self.user)
+
+        self.project = Project.objects.create(
+            name="Job Permission Project", description="For testing job run permission"
+        )
+        assign_perm(Project.Permissions.CREATE_JOB, self.user, self.project)
+        self.valid_job_keys = [cls.key for cls in VALID_JOB_TYPES if cls.key != "unknown"]
+
+    def _create_job(self, job_type_key):
+        job = Job.objects.create(name="Test Job", project=self.project, job_type_key=job_type_key)
+        return job
+
+    def assign_run_permission(self, key):
+        perm = f"main.run_{key}_job"
+        assign_perm(perm, self.user, self.project)
+
+    def remove_run_permission(self, key):
+        perm = f"main.run_{key}_job"
+        remove_perm(perm, self.user, self.project)
+
+    def test_can_only_run_permitted_job_type(self):
+        allowed_key = self.valid_job_keys[0]
+        self.assign_run_permission(allowed_key)
+
+        for job_type_key in self.valid_job_keys:
+            job = self._create_job(job_type_key)
+            response = self.client.post(f"/api/v2/jobs/{job.pk}/run/", format="json")
+            if job_type_key == allowed_key:
+                self.assertEqual(response.status_code, status.HTTP_200_OK, f"{job_type_key} should run successfully")
+            else:
+                self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN, f"{job_type_key} should be denied")
+
+    def test_can_run_multiple_if_permitted(self):
+        allowed_keys = self.valid_job_keys[:2]
+        for key in allowed_keys:
+            self.assign_run_permission(key)
+
+        for job_type_key in self.valid_job_keys:
+            job = self._create_job(job_type_key)
+            response = self.client.post(f"/api/v2/jobs/{job.pk}/run/", format="json")
+
+            if job_type_key in allowed_keys:
+                self.assertEqual(response.status_code, status.HTTP_200_OK, f"{job_type_key} should run successfully")
+            else:
+                self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN, f"{job_type_key} should be denied")
+
+    def test_cannot_run_any_without_permission(self):
+        for job_type_key in self.valid_job_keys:
+            job = self._create_job(job_type_key)
+            response = self.client.post(f"/api/v2/jobs/{job.pk}/run/", format="json")
+            self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN, f"{job_type_key} should be denied")
+
+    def test_user_permissions_reflected_in_job_detail(self):
+        job = self._create_job(job_type_key="ml")
+
+        # By default, the user shouldn't have any job-related perms
+        response = self.client.get(f"/api/v2/jobs/{job.pk}/")
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.data.get("user_permissions"), [], "User should not have any job perms initially")
+
+        # Assign run permission and check if it's reflected
+        assign_perm(Project.Permissions.RUN_ML_JOB, self.user, self.project)
+        response = self.client.get(f"/api/v2/jobs/{job.pk}/")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("run", response.data.get("user_permissions", []))
+
+        # Remove run permission and confirm it's removed
+        remove_perm(Project.Permissions.RUN_ML_JOB, self.user, self.project)
+        response = self.client.get(f"/api/v2/jobs/{job.pk}/")
+        self.assertEqual(response.status_code, 200)
+        self.assertNotIn("run", response.data.get("user_permissions", []))
+
+
+class TestRunSingleImageJobPermission(APITestCase):
+    def setUp(self):
+        super().setUp()
+        self.user = User.objects.create_user(
+            email="regularuser@insectai.org",
+            password="password123",
+        )
+        self.project = Project.objects.create(name="Single Image Project", description="Testing single image job")
+        self.pipeline = Pipeline.objects.create(
+            name="Test ML pipeline",
+            description="Test ML pipeline",
+        )
+        self.pipeline.projects.add(self.project)
+        self.deployment = Deployment.objects.create(name="Test Deployment", project=self.project)
+        create_captures(deployment=self.deployment)
+        group_images_into_events(deployment=self.deployment)
+        self.capture = self.deployment.captures.first()
+        self.client.force_authenticate(self.user)
+
+    def _grant_create_job__and_run_single_image_perm(self):
+        """Grants run_single_image_job permission on a specific capture to the user."""
+        assign_perm(Project.Permissions.CREATE_JOB, self.user, self.project)
+        assign_perm(Project.Permissions.RUN_SINGLE_IMAGE_JOB, self.user, self.project)
+
+    def _remove_run_single_image_perm(self):
+        remove_perm(Project.Permissions.RUN_SINGLE_IMAGE_JOB, self.user, self.project)
+
+    def test_user_can_run_single_image_job_and_perm_is_reflected(self):
+        self._grant_create_job__and_run_single_image_perm()
+
+        # Verify permission is reflected in capture detail response
+        capture_detail_url = f"/api/v2/captures/{self.capture.pk}/"
+        response = self.client.get(capture_detail_url)
+        self.assertEqual(response.status_code, 200)
+        self.assertIn(
+            "run_single_image_ml_job",
+            response.data.get("user_permissions", []),
+            "run_single_image permission not reflected",
+        )
+
+        # Try to run a job using source_image_single_id
+        run_url = "/api/v2/jobs/?start_now"
+        payload = {
+            "delay": 0,
+            "name": f"Capture #{self.capture.pk}",
+            "project_id": str(self.project.pk),
+            "pipeline_id": str(self.pipeline.pk),
+            "source_image_single_id": str(self.capture.pk),
+        }
+        response = self.client.post(run_url, payload, format="json")
+        self.assertEqual(
+            response.status_code, 201, f"User should be able to run single image job, got {response.status_code}"
+        )
+        # Remove permission
+        self._remove_run_single_image_perm()
+
+        # Permission should no longer appear in capture detail
+        response = self.client.get(capture_detail_url)
+        self.assertEqual(response.status_code, 200)
+        self.assertNotIn(
+            "run_single_image_ml_job",
+            response.data.get("user_permissions", []),
+            "run_single_image permission should be removed but still present",
+        )
+
+        # Should not be able to run job now
+        response = self.client.post(run_url, payload, format="json")
+        self.assertEqual(
+            response.status_code,
+            403,
+            f"User should NOT be able to run single image job after permission removal, got {response.status_code}",
+        )
diff --git a/ami/users/roles.py b/ami/users/roles.py
index c14ad0f76..a3c85c3c5 100644
--- a/ami/users/roles.py
+++ b/ami/users/roles.py
@@ -56,6 +56,8 @@ class BasicMember(Role):
     permissions = Role.permissions | {
         Project.Permissions.VIEW_PRIVATE_DATA,
         Project.Permissions.STAR_SOURCE_IMAGE,
+        Project.Permissions.CREATE_JOB,
+        Project.Permissions.RUN_SINGLE_IMAGE_JOB,
     }
 
 
@@ -75,9 +77,11 @@ class MLDataManager(Role):
     permissions = BasicMember.permissions | {
         Project.Permissions.CREATE_JOB,
         Project.Permissions.UPDATE_JOB,
-        Project.Permissions.RUN_JOB,
-        Project.Permissions.RETRY_JOB,
-        Project.Permissions.CANCEL_JOB,
+        # RUN ML jobs is revoked for now
+        # Project.Permissions.RUN_ML_JOB,
+        Project.Permissions.RUN_POPULATE_CAPTURES_COLLECTION_JOB,
+        Project.Permissions.RUN_DATA_STORAGE_SYNC_JOB,
+        Project.Permissions.RUN_DATA_EXPORT_JOB,
         Project.Permissions.DELETE_JOB,
         Project.Permissions.DELETE_OCCURRENCES,
     }
@@ -90,7 +94,6 @@ class ProjectManager(Role):
         | Identifier.permissions
         | MLDataManager.permissions
         | {
-            Project.Permissions.CHANGE,
             Project.Permissions.CHANGE,
             Project.Permissions.DELETE,
             Project.Permissions.IMPORT_DATA,
diff --git a/ui/src/data-services/models/capture-details.ts b/ui/src/data-services/models/capture-details.ts
index b88b3f796..c1c83c301 100644
--- a/ui/src/data-services/models/capture-details.ts
+++ b/ui/src/data-services/models/capture-details.ts
@@ -1,3 +1,4 @@
+import { UserPermission } from 'utils/user/types'
 import { Capture, ServerCapture } from './capture'
 import { Job } from './job'
 
@@ -69,4 +70,8 @@ export class CaptureDetails extends Capture {
   get url(): string {
     return this._capture.url
   }
+
+  get userPermissions(): UserPermission[] {
+    return this._capture.user_permissions
+  }
 }
diff --git a/ui/src/data-services/models/job.ts b/ui/src/data-services/models/job.ts
index 120d1efb1..625db4ec7 100644
--- a/ui/src/data-services/models/job.ts
+++ b/ui/src/data-services/models/job.ts
@@ -45,7 +45,7 @@ export class Job {
 
   get canCancel(): boolean {
     return (
-      this._job.user_permissions.includes(UserPermission.Cancel) &&
+      this._job.user_permissions.includes(UserPermission.Run) &&
       (this.status.code === 'STARTED' || this.status.code === 'PENDING')
     )
   }
@@ -63,7 +63,7 @@ export class Job {
 
   get canRetry(): boolean {
     return (
-      this._job.user_permissions.includes(UserPermission.Retry) &&
+      this._job.user_permissions.includes(UserPermission.Run) &&
       this.status.code !== 'CREATED' &&
       this.status.code !== 'STARTED' &&
       this.status.code !== 'PENDING'
diff --git a/ui/src/pages/session-details/playback/capture-details/capture-job/process-now.tsx b/ui/src/pages/session-details/playback/capture-details/capture-job/process-now.tsx
index dca941780..3ccbc885b 100644
--- a/ui/src/pages/session-details/playback/capture-details/capture-job/process-now.tsx
+++ b/ui/src/pages/session-details/playback/capture-details/capture-job/process-now.tsx
@@ -1,11 +1,11 @@
 import { useCreateJob } from 'data-services/hooks/jobs/useCreateJob'
-import { useProjectDetails } from 'data-services/hooks/projects/useProjectDetails'
 import { CaptureDetails } from 'data-services/models/capture-details'
 import { Tooltip } from 'design-system/components/tooltip/tooltip'
 import { CheckIcon, Loader2Icon } from 'lucide-react'
 import { Button } from 'nova-ui-kit'
 import { useParams } from 'react-router-dom'
 import { STRING, translate } from 'utils/language'
+import { UserPermission } from 'utils/user/types'
 
 export const ProcessNow = ({
   capture,
@@ -15,14 +15,16 @@ export const ProcessNow = ({
   pipelineId?: string
 }) => {
   const { projectId } = useParams()
-  const { project } = useProjectDetails(projectId as string, true)
   const { createJob, isLoading, isSuccess } = useCreateJob()
+  const canProcess = capture?.userPermissions.includes(
+    UserPermission.RunSingleImage
+  )
   const disabled =
-    !capture || capture.hasJobInProgress || !pipelineId || !project?.canUpdate
+    !capture || capture.hasJobInProgress || !pipelineId || !canProcess
 
   // @TODO: hasJobInProgress, replace with if pipeline is healthy/available
 
-  const tooltipContent = project?.canUpdate
+  const tooltipContent = canProcess
     ? translate(STRING.MESSAGE_PROCESS_NOW_TOOLTIP)
     : translate(STRING.MESSAGE_PERMISSIONS_MISSING)
 
diff --git a/ui/src/utils/user/types.ts b/ui/src/utils/user/types.ts
index 35fa199a7..1a6d5833e 100644
--- a/ui/src/utils/user/types.ts
+++ b/ui/src/utils/user/types.ts
@@ -11,14 +11,13 @@ export type UserInfo = {
 }
 
 export enum UserPermission {
-  Cancel = 'cancel', // Custom job permission
   Create = 'create',
   Delete = 'delete',
   Populate = 'populate', // Custom collection permission
-  Retry = 'retry', // Custom job permission
   Run = 'run', // Custom job permission
+  RunSingleImage = 'run_single_image_ml_job', // Custom job permission
   Star = 'star',
-  Update = 'update', // Custom capture permission
+  Update = 'update',
 }
 
 export interface UserContextValues {