feat(perms,tasks): regroup_sessions_deployment perm + idempotent regroup task

Addresses takeaway-review feedback on PR #1292:

- Add `regroup_sessions_deployment` custom Project permission so non-superusers
  with the ProjectManager role can hit `POST /deployments/<pk>/regroup-sessions/`
  (mirrors `sync_deployment`). Without this, `BaseModel.check_custom_permission`
  computes the codename `regroup_sessions_deployment` and the perm doesn't
  exist, so the action 403s for everyone except superusers.
- Wrap `ami.tasks.regroup_events` in a per-deployment cache lock. Concurrent
  enqueues (double-clicks, sync_captures auto-regroup racing with manual
  trigger) collapse to a single run instead of overlapping on the same rows.
  Lower-level than the view so all callers (Deployment.save, admin bulk
  action, new API action) are protected.
- Defensive clamp on `session_time_gap_seconds` — fall back to the historical
  120-minute default if the value is 0 / negative, with a warning log. Stops
  pathological "every gap = new event" behavior when admins enter bad values.
- Migration 0085 adds the new perm to `Project.Meta.permissions` and grants it
  to existing `ProjectManager` role groups via a data migration (mirrors how
  0084 revoked `delete_job`).
- Tests:
  - test_invalid_session_time_gap_falls_back_to_default exercises 0 / -1 / -7200
  - test_regroup_events_task_is_idempotent_under_concurrent_calls pre-takes
    the cache lock, asserts a second call skips group_images_into_events

Co-Authored-By: Claude <noreply@anthropic.com>

Author: mihow

ami/main/migrations/0085_add_regroup_sessions_deployment_permission.py

@@ -0,0 +1,135 @@
+"""
+Add the ``regroup_sessions_deployment`` custom permission to ``Project`` and
+grant it to existing ``ProjectManager`` role groups.
+
+The new ``POST /api/v2/deployments/<pk>/regroup-sessions/`` action runs through
+``BaseModel.check_custom_permission``, which builds the codename as
+``{action}_{model_name}`` — for the ``regroup_sessions`` action on a
+``Deployment`` viewset that resolves project permission via the parent project,
+the perm needed is ``regroup_sessions_deployment``. Mirrors how
+``sync_deployment`` is granted in ``ami.users.roles.ProjectManager``.
+"""
+
+from django.db import migrations
+from django.db.models import Q
+
+
+def grant_regroup_sessions_to_project_managers(apps, schema_editor):
+    Group = apps.get_model("auth", "Group")
+    Permission = apps.get_model("auth", "Permission")
+    ContentType = apps.get_model("contenttypes", "ContentType")
+
+    try:
+        project_ct = ContentType.objects.get(app_label="main", model="project")
+    except ContentType.DoesNotExist:
+        return
+
+    perm, _ = Permission.objects.get_or_create(
+        codename="regroup_sessions_deployment",
+        content_type=project_ct,
+        defaults={"name": "Can regroup deployment captures into sessions"},
+    )
+
+    role_groups = Group.objects.filter(Q(name__endswith="_ProjectManager"))
+    for group in role_groups:
+        group.permissions.add(perm)
+
+
+def revoke_regroup_sessions_from_project_managers(apps, schema_editor):
+    Group = apps.get_model("auth", "Group")
+    Permission = apps.get_model("auth", "Permission")
+    ContentType = apps.get_model("contenttypes", "ContentType")
+    GroupObjectPermission = apps.get_model("guardian", "GroupObjectPermission")
+
+    try:
+        project_ct = ContentType.objects.get(app_label="main", model="project")
+    except ContentType.DoesNotExist:
+        return
+    try:
+        perm = Permission.objects.get(codename="regroup_sessions_deployment", content_type=project_ct)
+    except Permission.DoesNotExist:
+        return
+
+    role_groups = Group.objects.filter(Q(name__endswith="_ProjectManager"))
+    for group in role_groups:
+        group.permissions.remove(perm)
+
+    GroupObjectPermission.objects.filter(
+        permission=perm,
+        content_type=project_ct,
+        group__in=role_groups,
+    ).delete()
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("main", "0084_revoke_delete_job_from_roles"),
+        ("guardian", "0002_generic_permissions_index"),
+    ]
+
+    operations = [
+        migrations.AlterModelOptions(
+            name="project",
+            options={
+                "ordering": ["-priority", "created_at"],
+                "permissions": [
+                    ("create_identification", "Can create identifications"),
+                    ("update_identification", "Can update identifications"),
+                    ("delete_identification", "Can delete identifications"),
+                    ("create_job", "Can create a job"),
+                    ("update_job", "Can update 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"),
+                    ("run_post_processing_job", "Can run/retry/cancel Post-Processing jobs"),
+                    ("delete_job", "Can delete a job"),
+                    ("create_deployment", "Can create a deployment"),
+                    ("delete_deployment", "Can delete a deployment"),
+                    ("update_deployment", "Can update a deployment"),
+                    ("sync_deployment", "Can sync images to a deployment"),
+                    ("regroup_sessions_deployment", "Can regroup deployment captures into sessions"),
+                    ("create_sourceimagecollection", "Can create a collection"),
+                    ("update_sourceimagecollection", "Can update a collection"),
+                    ("delete_sourceimagecollection", "Can delete a collection"),
+                    ("populate_sourceimagecollection", "Can populate a collection"),
+                    ("create_sourceimage", "Can create a source image"),
+                    ("update_sourceimage", "Can update a source image"),
+                    ("delete_sourceimage", "Can delete a source image"),
+                    ("star_sourceimage", "Can star a source image"),
+                    ("create_sourceimageupload", "Can create a source image upload"),
+                    ("update_sourceimageupload", "Can update a source image upload"),
+                    ("delete_sourceimageupload", "Can delete a source image upload"),
+                    ("create_s3storagesource", "Can create storage"),
+                    ("delete_s3storagesource", "Can delete storage"),
+                    ("update_s3storagesource", "Can update storage"),
+                    ("test_s3storagesource", "Can test storage connection"),
+                    ("create_site", "Can create a site"),
+                    ("delete_site", "Can delete a site"),
+                    ("update_site", "Can update a site"),
+                    ("create_device", "Can create a device"),
+                    ("delete_device", "Can delete a device"),
+                    ("update_device", "Can update a device"),
+                    ("view_userprojectmembership", "Can view project members"),
+                    ("create_userprojectmembership", "Can add a user to the project"),
+                    ("update_userprojectmembership", "Can update a user's project membership and role in the project"),
+                    ("delete_userprojectmembership", "Can remove a user from the project"),
+                    ("create_dataexport", "Can create a data export"),
+                    ("update_dataexport", "Can update a data export"),
+                    ("delete_dataexport", "Can delete a data export"),
+                    ("create_projectpipelineconfig", "Can register pipelines for the project"),
+                    ("update_projectpipelineconfig", "Can update pipeline configurations"),
+                    ("delete_projectpipelineconfig", "Can remove pipelines from the project"),
+                    ("create_taxalist", "Can create a taxa list"),
+                    ("update_taxalist", "Can update a taxa list"),
+                    ("delete_taxalist", "Can delete a taxa list"),
+                    ("view_private_data", "Can view private data"),
+                ],
+            },
+        ),
+        migrations.RunPython(
+            grant_regroup_sessions_to_project_managers,
+            revoke_regroup_sessions_from_project_managers,
+        ),
+    ]

ami/main/models.py

@@ -419,6 +419,7 @@ class Permissions:
         DELETE_DEPLOYMENT = "delete_deployment"
         UPDATE_DEPLOYMENT = "update_deployment"
         SYNC_DEPLOYMENT = "sync_deployment"
+        REGROUP_SESSIONS_DEPLOYMENT = "regroup_sessions_deployment"
 
         # Collection permissions
         CREATE_COLLECTION = "create_sourceimagecollection"
@@ -499,6 +500,7 @@ class Meta:
             ("delete_deployment", "Can delete a deployment"),
             ("update_deployment", "Can update a deployment"),
             ("sync_deployment", "Can sync images to a deployment"),
+            ("regroup_sessions_deployment", "Can regroup deployment captures into sessions"),
             # Collection permissions
             ("create_sourceimagecollection", "Can create a collection"),
             ("update_sourceimagecollection", "Can update a collection"),
@@ -1399,10 +1401,19 @@ def group_images_into_events(
     max_event_duration: datetime.timedelta | None = DEFAULT_MAX_EVENT_DURATION,
 ) -> list[Event]:
     if max_time_gap is None:
+        default_gap = datetime.timedelta(minutes=120)
         if deployment.project_id:
-            max_time_gap = datetime.timedelta(seconds=deployment.project.session_time_gap_seconds)
+            gap_seconds = deployment.project.session_time_gap_seconds
+            if gap_seconds is None or gap_seconds <= 0:
+                logger.warning(
+                    f"Project {deployment.project_id} has invalid session_time_gap_seconds "
+                    f"({gap_seconds!r}); falling back to default {default_gap}"
+                )
+                max_time_gap = default_gap
+            else:
+                max_time_gap = datetime.timedelta(seconds=gap_seconds)
         else:
-            max_time_gap = datetime.timedelta(minutes=120)
+            max_time_gap = default_gap
     # Log a warning if multiple SourceImages have the same timestamp
     dupes = (
         SourceImage.objects.filter(deployment=deployment)

ami/main/tests.py

@@ -525,6 +525,60 @@ def test_session_time_gap_seconds_is_used_when_no_explicit_gap(self):
         count_4h = Event.objects.filter(deployment=self.deployment).count()
         assert count_4h == 1, f"expected 1 Event at project setting 14400s, got {count_4h}"
 
+    def test_invalid_session_time_gap_falls_back_to_default(self):
+        """
+        Non-positive (0 or negative) ``session_time_gap_seconds`` would
+        otherwise split every timestamp into its own Event. Guard by falling
+        back to the historical 120-minute default and logging a warning.
+        """
+        self._create_burst(datetime.datetime(2023, 8, 5, 22, 0), n=6, interval_minutes=5)
+        self._create_burst(datetime.datetime(2023, 8, 6, 1, 0), n=6, interval_minutes=5)
+
+        for bad_value in (0, -1, -7200):
+            Event.objects.filter(deployment=self.deployment).delete()
+            self.project.session_time_gap_seconds = bad_value
+            self.project.save()
+            self.deployment.refresh_from_db()
+            group_images_into_events(deployment=self.deployment)
+            count = Event.objects.filter(deployment=self.deployment).count()
+            # Default 120-min gap on this cross-midnight pattern → 2 Events,
+            # NOT 12 (which is what bad_value=0 would produce without the guard).
+            assert count == 2, f"expected 2 Events at gap={bad_value} (default fallback), got {count}"
+
+    def test_regroup_events_task_is_idempotent_under_concurrent_calls(self):
+        """
+        ``ami.tasks.regroup_events`` uses a per-deployment cache lock so
+        concurrent enqueues collapse to a single run. The second call should
+        skip without touching the DB.
+        """
+        from unittest.mock import patch
+
+        from django.core.cache import cache
+
+        from ami.tasks import regroup_events
+
+        # Make sure no stale lock from a prior test leaks in.
+        cache.delete(f"regroup_events:lock:deployment:{self.deployment.pk}")
+
+        self._create_burst(datetime.datetime(2023, 8, 5, 22, 0), n=6, interval_minutes=5)
+
+        with patch("ami.main.models.group_images_into_events") as mock_group:
+            # Pre-take the lock to simulate an in-flight run.
+            cache.add(f"regroup_events:lock:deployment:{self.deployment.pk}", 1, timeout=60)
+            try:
+                regroup_events(self.deployment.pk)
+            finally:
+                cache.delete(f"regroup_events:lock:deployment:{self.deployment.pk}")
+            assert mock_group.call_count == 0, "Locked run should skip group_images_into_events"
+
+        # Without the pre-taken lock, the next call should run normally and release the lock.
+        with patch("ami.main.models.group_images_into_events", return_value=[]) as mock_group:
+            regroup_events(self.deployment.pk)
+            assert mock_group.call_count == 1, "Unlocked run should invoke group_images_into_events"
+        assert (
+            cache.get(f"regroup_events:lock:deployment:{self.deployment.pk}") is None
+        ), "Lock should be released after the task body completes"
+
     def test_pruning_empty_events(self):
         from ami.main.models import delete_empty_events
 

ami/tasks.py

@@ -1,6 +1,7 @@
 import logging
 
 from django.apps import apps
+from django.core.cache import cache
 from django.db import models
 
 from config import celery_app
@@ -13,6 +14,11 @@
 default_time_limit = two_days + one_hour
 default_soft_time_limit = two_days
 
+# Lock TTL for regroup_events. Matches the task's soft_time_limit so a stuck
+# task cannot wedge the lock indefinitely; under normal load regroup completes
+# in seconds-to-minutes, well below this ceiling.
+REGROUP_EVENTS_LOCK_TTL = one_hour
+
 
 # @TODO use shared_task decorator instead of celery_app?
 @celery_app.task(soft_time_limit=two_days, time_limit=two_days + one_hour)
@@ -91,14 +97,26 @@ def populate_collection(collection_id: int) -> None:
 def regroup_events(deployment_id: int) -> None:
     from ami.main.models import Deployment, group_images_into_events
 
-    try:
-        deployment = Deployment.objects.get(id=deployment_id)
-    except Deployment.DoesNotExist:
-        logger.error(f"Deployment with id {deployment_id} not found")
+    # Per-deployment idempotency lock. Concurrent enqueues (e.g. user double-
+    # clicks the regroup button, sync_captures auto-regroup races with manual
+    # admin trigger) collapse to a single run. The cache.add() is atomic
+    # SETNX-style: only the first caller takes the lock; later callers exit
+    # early without touching the DB.
+    lock_key = f"regroup_events:lock:deployment:{deployment_id}"
+    if not cache.add(lock_key, 1, timeout=REGROUP_EVENTS_LOCK_TTL):
+        logger.info(f"regroup_events skipped for deployment {deployment_id}: another run is in progress.")
         return
-    logger.info(f"Grouping captures for {deployment}")
-    events = group_images_into_events(deployment)
-    logger.info(f"{deployment} now has {len(events)} events")
+    try:
+        try:
+            deployment = Deployment.objects.get(id=deployment_id)
+        except Deployment.DoesNotExist:
+            logger.error(f"Deployment with id {deployment_id} not found")
+            return
+        logger.info(f"Grouping captures for {deployment}")
+        events = group_images_into_events(deployment)
+        logger.info(f"{deployment} now has {len(events)} events")
+    finally:
+        cache.delete(lock_key)
 
 
 @celery_app.task(soft_time_limit=one_hour, time_limit=one_hour + 60)

ami/users/roles.py

@@ -181,6 +181,7 @@ class ProjectManager(Role):
             Project.Permissions.UPDATE_DEPLOYMENT,
             Project.Permissions.DELETE_DEPLOYMENT,
             Project.Permissions.SYNC_DEPLOYMENT,
+            Project.Permissions.REGROUP_SESSIONS_DEPLOYMENT,
             Project.Permissions.CREATE_SITE,
             Project.Permissions.UPDATE_SITE,
             Project.Permissions.DELETE_SITE,