diff --git a/ami/jobs/management/commands/test_regroup_job_e2e.py b/ami/jobs/management/commands/test_regroup_job_e2e.py
new file mode 100644
index 000000000..6541e54a9
--- /dev/null
+++ b/ami/jobs/management/commands/test_regroup_job_e2e.py
@@ -0,0 +1,209 @@
+"""End-to-end test harness for ``RegroupEventsJob`` and the regroup stage of
+``DataStorageSyncJob``.
+
+Mirrors ``test_ml_job_e2e`` for the session-regrouping path (PR #1292). Used to
+validate:
+
+* Mode ``regroup``: ``RegroupEventsJob`` runs to SUCCESS against a real
+  deployment, stage params are populated, Event count delta is reasonable.
+* Mode ``sync``: ``DataStorageSyncJob`` exposes a two-stage progress
+  (sync_captures + regroup_sessions), both reach SUCCESS, regroup stage params
+  are populated.
+* Mode ``concurrent``: Two ``RegroupEventsJob`` enqueues for the same
+  deployment within the lock TTL — exactly one stage produces non-zero stats,
+  the other short-circuits with a lock warning, and Event count does not
+  diverge from the single-run baseline.
+"""
+
+import time
+
+from django.core.management.base import BaseCommand, CommandError
+
+from ami.jobs.models import DataStorageSyncJob, Job, JobState, RegroupEventsJob
+from ami.main.models import Deployment, Event
+
+
+class Command(BaseCommand):
+    help = (
+        "Run end-to-end tests for the regroup-events Job path.\n\n"
+        "Modes:\n"
+        "  regroup     — RegroupEventsJob on a deployment\n"
+        "  sync        — DataStorageSyncJob (covers sync→regroup chain)\n"
+        "  concurrent  — two RegroupEventsJobs back-to-back, asserts lock semantics\n"
+    )
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "mode",
+            choices=["regroup", "sync", "concurrent"],
+            help="Which scenario to exercise",
+        )
+        parser.add_argument("--deployment", type=int, required=True, help="Deployment ID")
+        parser.add_argument(
+            "--poll-interval", type=float, default=2.0, help="Seconds between Job state polls (default 2.0)"
+        )
+        parser.add_argument(
+            "--timeout", type=float, default=600.0, help="Max seconds to wait for each Job (default 600)"
+        )
+
+    def handle(self, *args, **options):
+        deployment = self._resolve_deployment(options["deployment"])
+        mode = options["mode"]
+
+        if mode == "regroup":
+            self._run_regroup(deployment, options)
+        elif mode == "sync":
+            self._run_sync(deployment, options)
+        elif mode == "concurrent":
+            self._run_concurrent(deployment, options)
+
+    def _resolve_deployment(self, deployment_id: int) -> Deployment:
+        try:
+            deployment = Deployment.objects.get(pk=deployment_id)
+        except Deployment.DoesNotExist:
+            raise CommandError(f"Deployment {deployment_id} not found")
+        self.stdout.write(
+            self.style.SUCCESS(
+                f"✓ Deployment {deployment.pk} '{deployment.name}' "
+                f"(project={deployment.project_id}, captures={deployment.captures_count})"
+            )
+        )
+        gap = deployment.project.session_time_gap_seconds if deployment.project_id else None
+        self.stdout.write(f"  project session_time_gap_seconds = {gap!r}")
+        before = Event.objects.filter(deployment=deployment).count()
+        self.stdout.write(f"  Events before run: {before}")
+        return deployment
+
+    def _run_regroup(self, deployment: Deployment, options: dict) -> None:
+        before = Event.objects.filter(deployment=deployment).count()
+        job = self._make_regroup_job(deployment, suffix="e2e-regroup")
+        self.stdout.write(f"\n🚀 RegroupEventsJob {job.pk} enqueueing")
+        job.enqueue()
+        self._monitor(job, options)
+        after = Event.objects.filter(deployment=deployment).count()
+        self.stdout.write(f"\nEvents after: {after} (Δ {after - before:+d})")
+        self._assert_status(job, expected=JobState.SUCCESS)
+        self._dump_stage_params(job)
+
+    def _run_sync(self, deployment: Deployment, options: dict) -> None:
+        if not deployment.data_source_id:
+            raise CommandError(
+                f"Deployment {deployment.pk} has no data_source — DataStorageSyncJob would fail immediately."
+            )
+        before = Event.objects.filter(deployment=deployment).count()
+        job = Job.objects.create(
+            name=f"E2E sync→regroup chain (deployment {deployment.pk})",
+            project=deployment.project,
+            deployment=deployment,
+            job_type_key=DataStorageSyncJob.key,
+        )
+        self.stdout.write(f"\n🚀 DataStorageSyncJob {job.pk} enqueueing")
+        job.enqueue()
+        self._monitor(job, options)
+        after = Event.objects.filter(deployment=deployment).count()
+        self.stdout.write(f"\nEvents after: {after} (Δ {after - before:+d})")
+        self._assert_status(job, expected=JobState.SUCCESS)
+
+        stage_keys = [s.key for s in (job.progress.stages or [])]
+        if DataStorageSyncJob.regroup_stage_key not in stage_keys:
+            raise CommandError(
+                f"❌ DataStorageSyncJob exposed stages {stage_keys!r} — missing "
+                f"'{DataStorageSyncJob.regroup_stage_key}' regroup stage."
+            )
+        self.stdout.write(self.style.SUCCESS(f"✓ Sync Job exposed both stages: {stage_keys!r}"))
+        self._dump_stage_params(job)
+
+    def _run_concurrent(self, deployment: Deployment, options: dict) -> None:
+        before = Event.objects.filter(deployment=deployment).count()
+        job_a = self._make_regroup_job(deployment, suffix="e2e-concurrent-A")
+        job_b = self._make_regroup_job(deployment, suffix="e2e-concurrent-B")
+        self.stdout.write(f"\n🚀 Enqueueing two RegroupEventsJobs back-to-back: {job_a.pk}, {job_b.pk}")
+        job_a.enqueue()
+        # No sleep between — we want both Celery tasks to race for the lock.
+        job_b.enqueue()
+
+        self.stdout.write("\nMonitoring job A:")
+        self._monitor(job_a, options)
+        self.stdout.write("\nMonitoring job B:")
+        self._monitor(job_b, options)
+
+        after = Event.objects.filter(deployment=deployment).count()
+        self.stdout.write(f"\nEvents after both jobs: {after} (Δ {after - before:+d})")
+
+        for job in (job_a, job_b):
+            self._assert_status(job, expected=JobState.SUCCESS)
+
+        params_a = self._stage_param_dict(job_a, RegroupEventsJob.key)
+        params_b = self._stage_param_dict(job_b, RegroupEventsJob.key)
+        self.stdout.write(f"\nJob A stage params: {params_a}")
+        self.stdout.write(f"Job B stage params: {params_b}")
+
+        # Exactly one of A/B should have done real work (captures_grouped > 0);
+        # the other should have short-circuited and reported the initial zeroes.
+        worked_a = (params_a.get("captures_grouped") or 0) > 0
+        worked_b = (params_b.get("captures_grouped") or 0) > 0
+        if worked_a == worked_b:
+            self.stdout.write(
+                self.style.WARNING(
+                    f"⚠ Lock did not separate runs as expected — both jobs reported "
+                    f"captures_grouped={params_a.get('captures_grouped')}/"
+                    f"{params_b.get('captures_grouped')}. "
+                    f"This can happen if the worker ran them serially fast enough that the lock cleared between."
+                )
+            )
+        else:
+            winner = "A" if worked_a else "B"
+            loser = "B" if worked_a else "A"
+            self.stdout.write(
+                self.style.SUCCESS(f"✓ Lock semantics held: job {winner} did the work, job {loser} short-circuited.")
+            )
+
+    def _make_regroup_job(self, deployment: Deployment, suffix: str) -> Job:
+        return Job.objects.create(
+            name=f"E2E {suffix} (deployment {deployment.pk})",
+            project=deployment.project,
+            deployment=deployment,
+            job_type_key=RegroupEventsJob.key,
+        )
+
+    def _monitor(self, job: Job, options: dict) -> None:
+        start = time.time()
+        timeout = options["timeout"]
+        interval = options["poll_interval"]
+        last_status = None
+        while True:
+            job.refresh_from_db()
+            elapsed = time.time() - start
+            if job.status != last_status:
+                self.stdout.write(f"  [{elapsed:6.1f}s] Job {job.pk} status: {job.status}")
+                last_status = job.status
+            if job.status in JobState.final_states():
+                self.stdout.write(f"  [{elapsed:6.1f}s] Job {job.pk} reached final state {job.status}")
+                return
+            if elapsed > timeout:
+                raise CommandError(
+                    f"❌ Job {job.pk} did not reach a final state within {timeout}s (status={job.status})"
+                )
+            time.sleep(interval)
+
+    def _assert_status(self, job: Job, expected: str) -> None:
+        if job.status != expected:
+            raise CommandError(
+                f"❌ Job {job.pk} ended with status {job.status!r} (expected {expected!r}). "
+                f"Stages: {[(s.key, s.status, s.progress) for s in (job.progress.stages or [])]}"
+            )
+        self.stdout.write(self.style.SUCCESS(f"✓ Job {job.pk} ended {expected}"))
+
+    def _stage_param_dict(self, job: Job, stage_key: str) -> dict:
+        for stage in job.progress.stages or []:
+            if stage.key == stage_key:
+                return {param.key: param.value for param in (stage.params or [])}
+        return {}
+
+    def _dump_stage_params(self, job: Job) -> None:
+        for stage in job.progress.stages or []:
+            self.stdout.write(
+                f"\n  Stage '{stage.name}' ({stage.key}): status={stage.status} progress={stage.progress}"
+            )
+            for param in stage.params or []:
+                self.stdout.write(f"    {param.name} [{param.key}]: {param.value}")
diff --git a/ami/jobs/models.py b/ami/jobs/models.py
index 3669b65c4..5e10ac1f4 100644
--- a/ami/jobs/models.py
+++ b/ami/jobs/models.py
@@ -672,9 +672,37 @@ def process_images(cls, job, images):
         job.save()
 
 
+# Human-readable param names for the regroup stage, surfaced both on the
+# standalone RegroupEventsJob and on DataStorageSyncJob's regroup stage. The
+# stable retrieval key is the slugify(name) form (e.g. "captures-grouped"),
+# produced by JobProgress.make_key — see _REGROUP_STAGE_PARAM_KEYS in
+# ami.main.models for the corresponding kwarg map.
+REGROUP_STAGE_PARAM_NAMES = (
+    "Captures grouped",
+    "Events created",
+    "Events touched",
+    "Empty events deleted",
+    "Duplicate timestamps",
+    "Ungrouped captures",
+    "Captures missing timestamp",
+)
+
+
 class DataStorageSyncJob(JobType):
+    """
+    Sync captures from the deployment's data source, then regroup them into
+    sessions as a separate tracked stage.
+
+    The regroup stage runs inside this job (not via ``Deployment.save()``
+    autoregroup) so its logs land on the same Job row and a regroup failure
+    flips the Job to FAILURE. Previously a sync would silently succeed even
+    if the post-sync regroup raised — see #1157.
+    """
+
     name = "Data storage sync"
     key = "data_storage_sync"
+    regroup_stage_key = "regroup_sessions"
+    regroup_stage_name = "Regroup sessions"
 
     @classmethod
     def run(cls, job: "Job"):
@@ -683,10 +711,16 @@ def run(cls, job: "Job"):
 
         This is meant to be called by an async task, not directly.
         """
+        from ami.main.models import group_images_into_events
 
-        job.progress.add_stage(cls.name)
+        job.progress.add_stage(cls.name, key=cls.key)
         job.progress.add_stage_param(cls.key, "Total files", 0)
         job.progress.add_stage_param(cls.key, "Failed", 0)
+
+        job.progress.add_stage(cls.regroup_stage_name, key=cls.regroup_stage_key)
+        for param_name in REGROUP_STAGE_PARAM_NAMES:
+            job.progress.add_stage_param(cls.regroup_stage_key, param_name, 0)
+
         job.update_status(JobState.STARTED)
         job.started_at = datetime.datetime.now()
         job.finished_at = None
@@ -694,27 +728,46 @@ def run(cls, job: "Job"):
 
         if not job.deployment:
             raise ValueError("No deployment provided for data storage sync job")
-        else:
-            job.logger.info(f"Syncing captures for deployment {job.deployment}")
-            job.progress.update_stage(
-                cls.key,
-                status=JobState.STARTED,
-                progress=0,
-                total_files=0,
-            )
-            job.save()
 
-            job.deployment.sync_captures(job=job)
+        job.logger.info(f"Syncing captures for deployment {job.deployment}")
+        job.progress.update_stage(
+            cls.key,
+            status=JobState.STARTED,
+            progress=0,
+            total_files=0,
+        )
+        job.save()
+
+        job.deployment.sync_captures(job=job, regroup_after=False)
 
-            job.logger.info(f"Finished syncing captures for deployment {job.deployment}")
-            job.progress.update_stage(
-                cls.key,
-                status=JobState.SUCCESS,
-                progress=1,
+        job.logger.info(f"Finished syncing captures for deployment {job.deployment}")
+        job.progress.update_stage(cls.key, status=JobState.SUCCESS, progress=1)
+        job.save()
+
+        job.logger.info(f"Regrouping captures into sessions for deployment {job.deployment}")
+        job.progress.update_stage(cls.regroup_stage_key, status=JobState.STARTED, progress=0)
+        job.save()
+
+        events = group_images_into_events(job.deployment, job=job, stage_key=cls.regroup_stage_key)
+        job.logger.info(f"Deployment {job.deployment} now has {len(events)} events after sync regroup.")
+
+        # The lock-miss branch in group_images_into_events returns []. If we
+        # just synced new captures, that means those captures are now sitting
+        # ungrouped because a concurrent regroup held the lock. They will be
+        # picked up by the next Deployment.save autoregroup (e.g. the next
+        # sync) — but flag it loudly on this Job so an admin watching this run
+        # sees what happened rather than a silently-empty regroup stage.
+        sync_total_files = job.progress.get_stage_param(cls.key, "total_files").value or 0
+        if not events and sync_total_files:
+            job.logger.warning(
+                f"Sync added {sync_total_files} files but the regroup stage was skipped because "
+                f"another regroup is in progress for deployment {job.deployment.pk}. The new captures "
+                f"will be grouped by the next sync or save. If this keeps happening, check the Jobs "
+                f"list for a stuck regroup_events task."
             )
-            job.update_status(JobState.SUCCESS)
-            job.save()
 
+        job.progress.update_stage(cls.regroup_stage_key, status=JobState.SUCCESS, progress=1)
+        job.update_status(JobState.SUCCESS)
         job.finished_at = datetime.datetime.now()
         job.save()
 
@@ -834,6 +887,54 @@ def run(cls, job: "Job"):
         job.save()
 
 
+class RegroupEventsJob(JobType):
+    """
+    Regroup a deployment's captures into Events using the project's
+    ``session_time_gap_seconds`` setting.
+
+    Single-stage job: ``group_images_into_events`` is one mostly-atomic SQL
+    pass with no per-image Python loop, so we cannot report incremental %
+    progress meaningfully. Stage transitions are CREATED → STARTED (0%) →
+    SUCCESS/FAILURE (100%). Summary stats (events created/touched/deleted,
+    duplicates, ungrouped captures) are written to the stage params by
+    ``group_images_into_events`` itself before it returns. Closes #1157, #1158.
+
+    Scope: grouping only. Propagating ``project_id`` to children lives on
+    ``Deployment.save()`` via ``update_children()`` — save the deployment
+    after moving it to push the new ``project_id`` down. Bare
+    ``ami.tasks.regroup_events`` has the same scope.
+    """
+
+    name = "Regroup sessions"
+    key = "regroup_events"
+
+    @classmethod
+    def run(cls, job: "Job"):
+        from ami.main.models import group_images_into_events
+
+        if not job.deployment:
+            raise ValueError("No deployment provided for regroup events job")
+
+        job.progress.add_stage(cls.name, key=cls.key)
+        for param_name in REGROUP_STAGE_PARAM_NAMES:
+            job.progress.add_stage_param(cls.key, param_name, 0)
+
+        job.update_status(JobState.STARTED)
+        job.started_at = datetime.datetime.now()
+        job.finished_at = None
+        job.progress.update_stage(cls.key, status=JobState.STARTED, progress=0)
+        job.save()
+
+        job.logger.info(f"Regrouping captures for deployment {job.deployment}")
+        events = group_images_into_events(job.deployment, job=job, stage_key=cls.key)
+        job.logger.info(f"Deployment {job.deployment} now has {len(events)} events after regrouping.")
+
+        job.progress.update_stage(cls.key, status=JobState.SUCCESS, progress=1)
+        job.update_status(JobState.SUCCESS, save=False)
+        job.finished_at = datetime.datetime.now()
+        job.save()
+
+
 class UnknownJobType(JobType):
     name = "Unknown"
     key = "unknown"
@@ -847,6 +948,7 @@ def run(cls, job: "Job"):
     MLJob,
     SourceImageCollectionPopulateJob,
     DataStorageSyncJob,
+    RegroupEventsJob,
     UnknownJobType,
     DataExportJob,
     PostProcessingJob,
diff --git a/ami/jobs/tests/test_jobs.py b/ami/jobs/tests/test_jobs.py
index 783180ed7..7a7f8d5ff 100644
--- a/ami/jobs/tests/test_jobs.py
+++ b/ami/jobs/tests/test_jobs.py
@@ -9,18 +9,21 @@
 
 from ami.base.serializers import reverse_with_params
 from ami.jobs.models import (
+    DataStorageSyncJob,
     Job,
     JobDispatchMode,
     JobLog,
     JobProgress,
     JobState,
     MLJob,
+    RegroupEventsJob,
     SourceImageCollectionPopulateJob,
 )
-from ami.main.models import Project, SourceImage, SourceImageCollection
+from ami.main.models import Deployment, Event, Project, SourceImage, SourceImageCollection
 from ami.ml.models import Pipeline
 from ami.ml.models.processing_service import ProcessingService
 from ami.ml.orchestration.jobs import queue_images_to_nats
+from ami.tests.fixtures.main import create_captures
 from ami.users.models import User
 
 logger = logging.getLogger(__name__)
@@ -1539,3 +1542,119 @@ def test_task_updates_all_project_async_services(self):
         # may be set by the creation-time get_status() ping, but should be unchanged
         # after the task runs.
         self.assertEqual(sync_service.last_seen, sync_last_seen_before)
+
+
+class TestRegroupEventsJob(TestCase):
+    """
+    Cover the user-visible behaviour of the new RegroupEventsJob:
+    - Runs end-to-end against real captures, produces Events.
+    - Reports summary stats on the regroup stage.
+    - Surfaces a regroup failure as Job FAILURE so admins can see it in the UI.
+    """
+
+    def setUp(self):
+        super().setUp()
+        self.project = Project.objects.create(name="Regroup Job Project")
+        self.deployment = Deployment.objects.create(name="Regroup Job Deployment", project=self.project)
+        # Two nightly bursts → at least 2 Events expected
+        create_captures(deployment=self.deployment, num_nights=2, images_per_night=4, interval_minutes=2)
+
+    def test_regroup_job_runs_end_to_end_and_reports_stats(self):
+        Event.objects.filter(deployment=self.deployment).delete()
+        job = Job.objects.create(
+            name="Regroup test",
+            deployment=self.deployment,
+            project=self.project,
+            job_type_key=RegroupEventsJob.key,
+        )
+        job.run()
+        job.refresh_from_db()
+
+        self.assertEqual(job.status, JobState.SUCCESS.value)
+        # Stage and stats are populated.
+        stage = job.progress.get_stage(RegroupEventsJob.key)
+        self.assertEqual(stage.status, JobState.SUCCESS)
+        self.assertEqual(stage.progress, 1)
+        # Retrieval keys are slugify(name) with `-` → `_` (see python_slugify in
+        # ami.jobs.models). Names are the human labels shown in the Jobs UI.
+        params = {p.key: p.value for p in stage.params}
+        self.assertGreaterEqual(params["events_created"], 1, "should report at least one Event created")
+        self.assertGreaterEqual(params["captures_grouped"], 1)
+        names = {p.name for p in stage.params}
+        self.assertIn("Captures grouped", names)
+        self.assertIn("Events created", names)
+        # Real Events exist in the DB.
+        self.assertGreaterEqual(Event.objects.filter(deployment=self.deployment).count(), 1)
+
+    def test_regroup_job_failure_propagates(self):
+        """
+        When ``group_images_into_events`` raises, the JobType propagates so the
+        Celery wrapper records FAILURE. Verifies the regroup-after-sync silent-
+        failure pain (#1157) cannot happen in the Job path.
+        """
+        from unittest.mock import patch
+
+        job = Job.objects.create(
+            name="Regroup failure test",
+            deployment=self.deployment,
+            project=self.project,
+            job_type_key=RegroupEventsJob.key,
+        )
+        with patch("ami.main.models.group_images_into_events", side_effect=RuntimeError("boom")):
+            with self.assertRaises(RuntimeError):
+                job.run()
+
+
+class TestDataStorageSyncJobIncludesRegroupStage(TestCase):
+    """
+    The sync Job now runs grouping as an explicit second stage so logs land on
+    the same Job row and a regroup failure flips the Job to FAILURE rather than
+    silently succeeding (#1157).
+    """
+
+    def test_sync_job_adds_regroup_stage(self):
+        project = Project.objects.create(name="Sync Job Project")
+        deployment = Deployment.objects.create(name="Sync Job Deployment", project=project)
+        create_captures(deployment=deployment, num_nights=1, images_per_night=3, interval_minutes=2)
+
+        # Patch the inner sync call so we don't need S3; the grouping stage is what we're testing.
+        from unittest.mock import patch
+
+        with patch.object(Deployment, "sync_captures", return_value=3):
+            job = Job.objects.create(
+                name="Sync test",
+                deployment=deployment,
+                project=project,
+                job_type_key=DataStorageSyncJob.key,
+            )
+            job.run()
+            job.refresh_from_db()
+
+        self.assertEqual(job.status, JobState.SUCCESS.value)
+        regroup_stage = job.progress.get_stage(DataStorageSyncJob.regroup_stage_key)
+        self.assertEqual(regroup_stage.status, JobState.SUCCESS)
+        params = {p.key: p.value for p in regroup_stage.params}
+        self.assertIn("events_created", params)
+        self.assertIn("captures_grouped", params)
+        names = {p.name for p in regroup_stage.params}
+        self.assertIn("Captures grouped", names)
+        self.assertIn("Events created", names)
+
+    def test_sync_job_regroup_failure_propagates(self):
+        from unittest.mock import patch
+
+        project = Project.objects.create(name="Sync Job Failure Project")
+        deployment = Deployment.objects.create(name="Sync Job Failure Deployment", project=project)
+
+        job = Job.objects.create(
+            name="Sync regroup failure",
+            deployment=deployment,
+            project=project,
+            job_type_key=DataStorageSyncJob.key,
+        )
+        with patch.object(Deployment, "sync_captures", return_value=0), patch(
+            "ami.main.models.group_images_into_events",
+            side_effect=RuntimeError("regroup blew up"),
+        ):
+            with self.assertRaises(RuntimeError):
+                job.run()
diff --git a/ami/main/admin.py b/ami/main/admin.py
index ac9281ad4..f0a130577 100644
--- a/ami/main/admin.py
+++ b/ami/main/admin.py
@@ -1,3 +1,4 @@
+import datetime
 from typing import Any
 
 from django.contrib import admin
@@ -135,11 +136,11 @@ class DeploymentAdmin(admin.ModelAdmin[Deployment]):
         "name",
         "project",
         "data_source_uri",
-        "captures_count",
-        "captures_size",
-        "events_count",
-        "start_date",
-        "end_date",
+        "captures_count_display",
+        "captures_size_display",
+        "events_count_display",
+        "first_date",
+        "last_date",
     )
 
     search_fields = (
@@ -147,41 +148,83 @@ class DeploymentAdmin(admin.ModelAdmin[Deployment]):
         "name",
     )
 
-    def start_date(self, obj) -> str | None:
-        result = SourceImage.objects.filter(event__deployment=obj).aggregate(
-            models.Min("timestamp"),
-        )
-        return result["timestamp__min"].date() if result["timestamp__min"] else None
-
-    def end_date(self, obj) -> str | None:
-        result = SourceImage.objects.filter(deployment=obj).aggregate(
-            models.Max("timestamp"),
-        )
-        return result["timestamp__max"].date() if result["timestamp__max"] else None
-
-    def events_count(self, obj) -> str | None:
-        return number_format(obj.events.count(), force_grouping=True, use_l10n=True)
-
-    def captures_size(self, obj) -> str | None:
-        return filesizeformat(obj.data_source_total_size)
-
-    def captures_count(self, obj) -> str | None:
-        total_files = obj.data_source_total_files
-        return number_format(total_files, force_grouping=True, use_l10n=True)
+    # The previous custom start_date / end_date / events_count / captures_count
+    # methods each ran a live aggregate per row (`SourceImage.aggregate(Min/Max)`,
+    # `obj.events.count()`), making this list view ~O(rows × 4 expensive
+    # queries) — unusable on stacks with deployments holding >100k captures.
+    # Deployment already denormalizes captures_count, events_count,
+    # data_source_total_size, first_capture_timestamp and last_capture_timestamp
+    # (refreshed by update_calculated_fields); read those instead.
+    @admin.display(description="First date", ordering="first_capture_timestamp")
+    def first_date(self, obj: Deployment) -> datetime.date | None:
+        return obj.first_date()
+
+    @admin.display(description="Last date", ordering="last_capture_timestamp")
+    def last_date(self, obj: Deployment) -> datetime.date | None:
+        return obj.last_date()
+
+    @admin.display(description="Events", ordering="events_count")
+    def events_count_display(self, obj: Deployment) -> str:
+        return number_format(obj.events_count or 0, force_grouping=True, use_l10n=True)
+
+    @admin.display(description="Captures", ordering="captures_count")
+    def captures_count_display(self, obj: Deployment) -> str:
+        return number_format(obj.captures_count or 0, force_grouping=True, use_l10n=True)
+
+    @admin.display(description="Size", ordering="data_source_total_size")
+    def captures_size_display(self, obj: Deployment) -> str:
+        return filesizeformat(obj.data_source_total_size or 0)
 
     # list action that runs deployment.import_captures and displays a message
     # https://docs.djangoproject.com/en/3.2/ref/contrib/admin/actions/#writing-action-functions
-    @admin.action(description="Sync captures from deployment's data source (async)")
+    @admin.action(description="Sync captures from deployment's data source (Job)")
     def sync_captures(self, request: HttpRequest, queryset: QuerySet[Deployment]) -> None:
-        queued_tasks = [tasks.sync_source_images.delay(deployment.pk) for deployment in queryset]
-        msg = f"Syncing captures for {len(queued_tasks)} deployments in background: {queued_tasks}"
+        from ami.jobs.models import DataStorageSyncJob
+
+        queued_job_ids: list[int] = []
+        skipped: list[str] = []
+        for deployment in queryset:
+            if not deployment.project_id:
+                skipped.append(f"{deployment} (no project)")
+                continue
+            if not deployment.data_source_id:
+                skipped.append(f"{deployment} (no data source)")
+                continue
+            job = Job.objects.create(
+                name=f"Sync captures for deployment {deployment.pk}",
+                deployment=deployment,
+                project=deployment.project,
+                job_type_key=DataStorageSyncJob.key,
+            )
+            job.enqueue()
+            queued_job_ids.append(job.pk)
+        msg = f"Queued DataStorageSyncJob for {len(queued_job_ids)} deployments: {queued_job_ids}"
+        if skipped:
+            msg += f" — skipped: {', '.join(skipped)}"
         self.message_user(request, msg)
 
     # Action that regroups all captures in the deployment into events
-    @admin.action(description="Regroup captures into events (async)")
+    @admin.action(description="Regroup captures into sessions (Job)")
     def regroup_events(self, request: HttpRequest, queryset: QuerySet[Deployment]) -> None:
-        queued_tasks = [tasks.regroup_events.delay(deployment.pk) for deployment in queryset]
-        msg = f"Regrouping captures into events for {len(queued_tasks)} deployments in background: {queued_tasks}"
+        from ami.jobs.models import RegroupEventsJob
+
+        queued_job_ids: list[int] = []
+        skipped: list[str] = []
+        for deployment in queryset:
+            if not deployment.project_id:
+                skipped.append(f"{deployment} (no project)")
+                continue
+            job = Job.objects.create(
+                name=f"Regroup sessions for deployment {deployment.pk}",
+                deployment=deployment,
+                project=deployment.project,
+                job_type_key=RegroupEventsJob.key,
+            )
+            job.enqueue()
+            queued_job_ids.append(job.pk)
+        msg = f"Queued RegroupEventsJob for {len(queued_job_ids)} deployments: {queued_job_ids}"
+        if skipped:
+            msg += f" — skipped: {', '.join(skipped)}"
         self.message_user(request, msg)
 
     list_filter = ("project",)
diff --git a/ami/main/api/views.py b/ami/main/api/views.py
index 1bfa55d74..351ebd3ba 100644
--- a/ami/main/api/views.py
+++ b/ami/main/api/views.py
@@ -359,6 +359,45 @@ def sync(self, _request, pk=None) -> Response:
         else:
             raise api_exceptions.ValidationError(detail="Deployment must have a data source to sync captures from")
 
+    @action(detail=True, methods=["post"], name="regroup-sessions", url_path="regroup-sessions")
+    def regroup_sessions(self, _request, pk=None) -> Response:
+        """
+        Queue a ``RegroupEventsJob`` to regroup the deployment's source images into sessions.
+
+        Uses the project's ``session_time_gap_seconds`` setting to determine
+        the maximum gap between consecutive images before a new session is started.
+
+        (Sessions are stored as ``Event`` records internally.)
+        """
+        from ami.jobs.models import Job, RegroupEventsJob
+
+        deployment: Deployment = self.get_object()
+        if deployment.project_id is None:
+            # Schema allows it (project FK is nullable) but every Job carries a
+            # project and the regroup uses project.session_time_gap_seconds, so
+            # a project-less deployment can't run this endpoint.
+            raise api_exceptions.ValidationError(
+                detail={"deployment": "Deployment has no project; cannot enqueue regroup."}
+            )
+
+        job = Job.objects.create(
+            name=f"Regroup sessions for deployment {deployment.pk}",
+            deployment=deployment,
+            project=deployment.project,
+            job_type_key=RegroupEventsJob.key,
+        )
+        job.enqueue()
+        msg = f"Queued regroup sessions for deployment {deployment.pk} (job {job.pk})"
+        logger.info(msg)
+        return Response(
+            {
+                "job_id": job.pk,
+                "deployment_id": deployment.pk,
+                "project_id": deployment.project.pk,
+            },
+            status=status.HTTP_202_ACCEPTED,
+        )
+
     @extend_schema(parameters=[project_id_doc_param])
     def list(self, request, *args, **kwargs):
         return super().list(request, *args, **kwargs)
diff --git a/ami/main/migrations/0089_add_regroup_permissions.py b/ami/main/migrations/0089_add_regroup_permissions.py
new file mode 100644
index 000000000..8b898fb58
--- /dev/null
+++ b/ami/main/migrations/0089_add_regroup_permissions.py
@@ -0,0 +1,162 @@
+"""
+Add two custom permissions to ``Project`` and grant them to the relevant role
+groups:
+
+* ``regroup_sessions_deployment`` — gates the
+  ``POST /api/v2/deployments/<pk>/regroup-sessions/`` action via
+  ``BaseModel.check_custom_permission`` (codename = ``{action}_{model_name}``).
+  Mirrors how ``sync_deployment`` is granted. Goes to ``ProjectManager``.
+
+* ``run_regroup_events_job`` — gates create/run/cancel of the new
+  ``RegroupEventsJob`` type via ``Job.check_custom_permission`` (codename =
+  ``run_{job_type_key}_job``). Mirrors how ``run_data_storage_sync_job`` is
+  granted. Goes to ``MLDataManager`` (which ``ProjectManager`` inherits from).
+"""
+
+from django.db import migrations
+from django.db.models import Q
+
+
+def _get_or_create_perm(apps, codename: str, name: str):
+    Permission = apps.get_model("auth", "Permission")
+    ContentType = apps.get_model("contenttypes", "ContentType")
+    project_ct = ContentType.objects.get(app_label="main", model="project")
+    perm, _ = Permission.objects.get_or_create(
+        codename=codename,
+        content_type=project_ct,
+        defaults={"name": name},
+    )
+    return perm, project_ct
+
+
+def grant_new_permissions(apps, schema_editor):
+    Group = apps.get_model("auth", "Group")
+    ContentType = apps.get_model("contenttypes", "ContentType")
+
+    try:
+        ContentType.objects.get(app_label="main", model="project")
+    except ContentType.DoesNotExist:
+        return
+
+    regroup_perm, _ = _get_or_create_perm(
+        apps,
+        "regroup_sessions_deployment",
+        "Can regroup deployment captures into sessions",
+    )
+    run_regroup_job_perm, _ = _get_or_create_perm(
+        apps,
+        "run_regroup_events_job",
+        "Can run/retry/cancel Regroup Events jobs",
+    )
+
+    project_manager_groups = Group.objects.filter(Q(name__endswith="_ProjectManager"))
+    for group in project_manager_groups:
+        group.permissions.add(regroup_perm)
+
+    # run_regroup_events_job goes to anyone who can run a sync job today —
+    # MLDataManager is the primary holder; ProjectManager inherits from it
+    # and has its own role group.
+    job_runner_groups = Group.objects.filter(Q(name__endswith="_MLDataManager") | Q(name__endswith="_ProjectManager"))
+    for group in job_runner_groups:
+        group.permissions.add(run_regroup_job_perm)
+
+
+def revoke_new_permissions(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
+
+    for codename in ("regroup_sessions_deployment", "run_regroup_events_job"):
+        try:
+            perm = Permission.objects.get(codename=codename, content_type=project_ct)
+        except Permission.DoesNotExist:
+            continue
+        role_groups = Group.objects.filter(Q(name__endswith="_ProjectManager") | Q(name__endswith="_MLDataManager"))
+        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", "0088_detection_det_srcimg_created_idx"),
+        ("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_regroup_events_job", "Can run/retry/cancel Regroup Events 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_new_permissions,
+            revoke_new_permissions,
+        ),
+    ]
diff --git a/ami/main/migrations/0090_session_time_gap_seconds_help_text.py b/ami/main/migrations/0090_session_time_gap_seconds_help_text.py
new file mode 100644
index 000000000..9b369cdbd
--- /dev/null
+++ b/ami/main/migrations/0090_session_time_gap_seconds_help_text.py
@@ -0,0 +1,20 @@
+# Generated by Django 4.2.10 on 2026-06-08 17:33
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("main", "0089_add_regroup_permissions"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="project",
+            name="session_time_gap_seconds",
+            field=models.IntegerField(
+                default=7200,
+                help_text='Maximum gap between consecutive captures (in seconds) before a new session is started. Changing this value does not regroup existing captures automatically — to apply, select the affected deployments on the Deployments admin page and run the "Regroup captures into sessions" action.',
+            ),
+        ),
+    ]
diff --git a/ami/main/models.py b/ami/main/models.py
index 04210cbe6..b27874162 100644
--- a/ami/main/models.py
+++ b/ami/main/models.py
@@ -1,4 +1,5 @@
 import collections
+import contextlib
 import datetime
 import functools
 import logging
@@ -6,6 +7,7 @@
 import time
 import typing
 import urllib.parse
+import uuid
 from io import BytesIO
 from typing import Final, final  # noqa: F401
 
@@ -420,6 +422,7 @@ class Permissions:
         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_REGROUP_EVENTS_JOB = "run_regroup_events_job"
         RUN_DATA_EXPORT_JOB = "run_data_export_job"
         RUN_POST_PROCESSING_JOB = "run_post_processing_job"
         DELETE_JOB = "delete_job"
@@ -429,6 +432,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"
@@ -500,6 +504,7 @@ class Meta:
             ("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_regroup_events_job", "Can run/retry/cancel Regroup Events 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"),
@@ -509,6 +514,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"),
@@ -851,8 +857,20 @@ def data_source_total_size_display(self) -> str:
         else:
             return filesizeformat(self.data_source_total_size)
 
-    def sync_captures(self, batch_size=1000, regroup_events_per_batch=False, job: "Job | None" = None) -> int:
-        """Import images from the deployment's data source"""
+    def sync_captures(
+        self,
+        batch_size=1000,
+        regroup_events_per_batch=False,
+        regroup_after=True,
+        job: "Job | None" = None,
+    ) -> int:
+        """
+        Import images from the deployment's data source.
+
+        Set ``regroup_after=False`` when the caller (e.g. ``DataStorageSyncJob``)
+        will run regrouping as a tracked stage of its own so the work is visible
+        in the Jobs UI instead of buried inside ``Deployment.save()``.
+        """
 
         deployment = self
         assert deployment.data_source, f"Deployment {deployment.name} has no data source configured"
@@ -934,18 +952,29 @@ def sync_captures(self, batch_size=1000, regroup_events_per_batch=False, job: "J
 
         # @TODO decide if we should delete SourceImages that are no longer in the data source
 
-        if job:
-            job.logger.info("Saving and recalculating sessions for deployment")
-            job.progress.update_stage(job.job_type().key, progress=1)
-            job.progress.add_stage("Update deployment cache")
-            job.update_progress()
+        if regroup_after:
+            if job:
+                job.logger.info("Saving and recalculating sessions for deployment")
+                job.progress.update_stage(job.job_type().key, progress=1)
+                job.progress.add_stage("Update deployment cache")
+                job.update_progress()
 
-        # If new images were added, ensure the regroup happens now, not queued as an async task.
-        self.save(regroup_async=False)
+            # If new images were added, ensure the regroup happens now, not queued as an async task.
+            self.save(regroup_async=False)
 
-        if job:
-            job.progress.update_stage("Update deployment cache", progress=1)
-            job.update_progress()
+            if job:
+                job.progress.update_stage("Update deployment cache", progress=1)
+                job.update_progress()
+        else:
+            # Caller (e.g. DataStorageSyncJob) is responsible for running regroup as
+            # an explicit stage. Skip Deployment.save's autoregroup but still
+            # refresh cached counts and realign child Event/Occurrence/SourceImage
+            # project pointers — those normally run inside Deployment.save's
+            # update_calculated_fields branch, which we bypass here.
+            self.save(regroup_async=False, update_calculated_fields=False)
+            self.update_calculated_fields(save=True)
+            if self.project_id:
+                self.update_children()
 
         return total_files
 
@@ -1402,12 +1431,105 @@ def audit_event_lengths(deployment: Deployment):
 DEFAULT_MAX_EVENT_DURATION = datetime.timedelta(hours=24)
 
 
+# Regroup should finish in seconds, so keep this short: if a worker is killed
+# before its finally-block releases the lock, the TTL caps how long it stays held.
+# Matches the soft_time_limit on ami.tasks.regroup_events.
+REGROUP_LOCK_TTL_SECONDS = 10 * 60
+
+
+@contextlib.contextmanager
+def _regroup_lock(deployment_id: int):
+    """
+    Acquire a per-deployment lock for regrouping. Token-based release —
+    the lock is only deleted if the value we wrote is still there, so an
+    expired-then-reacquired-by-someone-else lock doesn't get clobbered.
+
+    Released cleanly on graceful exit and on any exception that propagates
+    out of the `with` block (including Celery's `SoftTimeLimitExceeded`).
+    Not released on hard worker death (OS SIGKILL, OOM-kill, pod eviction):
+    the token lives only in process memory, so a finally block that never
+    runs leaves the lock entry sitting until the TTL expires.
+
+    Yields True if acquired (caller should proceed), False if another run
+    already holds it (caller should short-circuit).
+    """
+    from django.core.cache import cache
+
+    lock_key = f"regroup_events:lock:deployment:{deployment_id}"
+    token = uuid.uuid4().hex
+    acquired = cache.add(lock_key, token, timeout=REGROUP_LOCK_TTL_SECONDS)
+    try:
+        yield acquired
+    finally:
+        if acquired:
+            current = cache.get(lock_key)
+            if current == token:
+                cache.delete(lock_key)
+
+
 def group_images_into_events(
     deployment: Deployment,
-    max_time_gap=datetime.timedelta(minutes=120),
+    max_time_gap: datetime.timedelta | None = None,
     delete_empty=True,
     max_event_duration: datetime.timedelta | None = DEFAULT_MAX_EVENT_DURATION,
+    job: "Job | None" = None,
+    stage_key: str | None = None,
 ) -> list[Event]:
+    """
+    Group a deployment's captures into Events based on timestamp gaps.
+
+    Holds a per-deployment cache lock so concurrent calls (autoregroup-on-save,
+    manual API/admin trigger, sync-time regroup) collapse to a single in-flight
+    run rather than racing on the same rows. If the lock is already held, this
+    function logs and returns an empty list without touching the DB.
+
+    When ``job`` and ``stage_key`` are passed, summary stats (events created,
+    events touched, duplicate-timestamp count, ungrouped captures) are written
+    to the named stage so the Jobs UI can surface them. Pure callers (e.g.
+    ``Deployment.save`` autoregroup, ``sync_captures`` per-batch) leave both
+    arguments at ``None``.
+    """
+    with _regroup_lock(deployment.pk) as acquired:
+        if not acquired:
+            msg = f"group_images_into_events skipped for deployment {deployment.pk}: another regroup is in progress."
+            if job:
+                job.logger.warning(msg)
+            else:
+                logger.warning(msg)
+            return []
+
+        return _group_images_into_events_locked(
+            deployment,
+            max_time_gap=max_time_gap,
+            delete_empty=delete_empty,
+            max_event_duration=max_event_duration,
+            job=job,
+            stage_key=stage_key,
+        )
+
+
+def _group_images_into_events_locked(
+    deployment: Deployment,
+    max_time_gap: datetime.timedelta | None,
+    delete_empty: bool,
+    max_event_duration: datetime.timedelta | None,
+    job: "Job | None",
+    stage_key: str | None,
+) -> list[Event]:
+    if max_time_gap is None:
+        default_gap = datetime.timedelta(minutes=120)
+        if deployment.project_id:
+            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 = default_gap
     # Log a warning if multiple SourceImages have the same timestamp
     dupes = (
         SourceImage.objects.filter(deployment=deployment)
@@ -1416,13 +1538,14 @@ def group_images_into_events(
         .filter(count__gt=1)
         .exclude(timestamp=None)
     )
-    if dupes.count():
-        values = "\n".join(
-            [f'{d.strftime("%Y-%m-%d %H:%M:%S")} x{c}' for d, c in dupes.values_list("timestamp", "count")]
+    duplicate_timestamp_count = dupes.count()
+    if duplicate_timestamp_count:
+        sample = "\n".join(
+            f'{d.strftime("%Y-%m-%d %H:%M:%S")} x{c}' for d, c in dupes.values_list("timestamp", "count")[:20]
         )
         logger.warning(
-            f"Found {len(values)} images with the same timestamp in deployment '{deployment}'. "
-            f"Only one image will be used for each timestamp for each event."
+            f"Found {duplicate_timestamp_count} duplicate-timestamp groups in deployment '{deployment}'. "
+            f"Only one image will be used per timestamp for each event. First 20:\n{sample}"
         )
 
     image_timestamps = list(
@@ -1440,6 +1563,7 @@ def group_images_into_events(
     )
 
     events = []
+    events_created_count = 0
     touched_event_pks: set[int] = set()
     for group in timestamp_groups:
         if not len(group):
@@ -1458,12 +1582,14 @@ def group_images_into_events(
 
         # Creating events & assigning images
         group_by = start_date.date()
-        event, _ = Event.objects.get_or_create(
+        event, was_created = Event.objects.get_or_create(
             deployment=deployment,
             group_by=group_by,
             defaults={"start": start_date, "end": end_date},
         )
         events.append(event)
+        if was_created:
+            events_created_count += 1
         touched_event_pks.add(event.pk)
 
         # Track events currently holding these captures — they'll lose captures
@@ -1522,9 +1648,10 @@ def group_images_into_events(
     if touched_event_pks:
         update_calculated_fields_for_events(pks=list(touched_event_pks))
 
+    events_deleted_empty = 0
     if delete_empty:
         logger.info("Deleting empty events for deployment")
-        delete_empty_events(deployment=deployment)
+        events_deleted_empty = delete_empty_events(deployment=deployment)
 
     for event in events:
         # Set the width and height of all images in each event based on the first image
@@ -1542,6 +1669,39 @@ def group_images_into_events(
 
     audit_event_lengths(deployment)
 
+    # Surface stats to the Jobs UI when this regroup is wrapped in a job.
+    # "Ungrouped captures" = images with valid timestamps that didn't land in
+    # any event (should be 0 unless an event delete races a capture insert;
+    # useful signal). Param NAMES below are the human-readable labels shown
+    # in the Jobs UI; stable retrieval keys are the slugify(name) forms (e.g.
+    # "captures-grouped"). See REGROUP_STAGE_PARAM_NAMES in ami.jobs.models.
+    if job and stage_key:
+        ungrouped_captures_count = (
+            SourceImage.objects.filter(deployment=deployment, event__isnull=True).exclude(timestamp=None).count()
+        )
+        no_timestamp_captures_count = SourceImage.objects.filter(deployment=deployment, timestamp=None).count()
+        # Count actual SourceImage rows assigned to an event, not the count of
+        # distinct timestamps in image_timestamps — two captures can share a
+        # timestamp (duplicate_timestamp_count tracks this) and the UPDATE above
+        # assigns both to the same event, so a distinct-timestamp count would
+        # understate the work done.
+        captures_grouped_count = (
+            SourceImage.objects.filter(deployment=deployment, event__isnull=False).exclude(timestamp=None).count()
+        )
+        regroup_stats = {
+            "Captures grouped": captures_grouped_count,
+            "Events created": events_created_count,
+            "Events touched": len(touched_event_pks),
+            "Empty events deleted": events_deleted_empty,
+            "Duplicate timestamps": duplicate_timestamp_count,
+            "Ungrouped captures": ungrouped_captures_count,
+            "Captures missing timestamp": no_timestamp_captures_count,
+        }
+        for param_name, value in regroup_stats.items():
+            job.progress.add_or_update_stage_param(stage_key, param_name, value)
+        job.update_progress()
+        job.save()
+
     return events
 
 
@@ -1589,9 +1749,11 @@ def deployment_events_need_update(deployment: Deployment) -> bool:
     return needs_update
 
 
-def delete_empty_events(deployment: Deployment, dry_run=False):
+def delete_empty_events(deployment: Deployment, dry_run=False) -> int:
     """
     Delete events that have no images, occurrences or other related records.
+
+    Returns the number of events deleted (or that would be deleted, for dry runs).
     """
 
     # @TODO Search all models that have a foreign key to Event
@@ -1610,12 +1772,14 @@ def delete_empty_events(deployment: Deployment, dry_run=False):
         .filter(num_images=0, num_occurrences=0)
     )
 
+    count = events.count()
     if dry_run:
         for event in events:
             logger.debug(f"Would delete event {event} (dry run)")
     else:
-        logger.info(f"Deleting {events.count()} empty events")
+        logger.info(f"Deleting {count} empty events")
         events.delete()
+    return count
 
 
 def sample_events(deployment: Deployment, day_interval: int = 3) -> typing.Generator[Event, None, None]:
diff --git a/ami/main/models_future/projects.py b/ami/main/models_future/projects.py
index 19b2156e8..0eb9c4a7f 100644
--- a/ami/main/models_future/projects.py
+++ b/ami/main/models_future/projects.py
@@ -26,7 +26,12 @@ class ProjectSettingsMixin(models.Model):
 
     session_time_gap_seconds = models.IntegerField(
         default=60 * 60 * 2,  # Default to 2 hours
-        help_text="Time gap in seconds to consider a new session",
+        help_text=(
+            "Maximum gap between consecutive captures (in seconds) before a new session is started. "
+            "Changing this value does not regroup existing captures automatically — to apply, "
+            "select the affected deployments on the Deployments admin page and run the "
+            '"Regroup captures into sessions" action.'
+        ),
     )
 
     default_filters_score_threshold = models.FloatField(
diff --git a/ami/main/tests.py b/ami/main/tests.py
index 5292a5f07..392ab2b15 100644
--- a/ami/main/tests.py
+++ b/ami/main/tests.py
@@ -440,6 +440,170 @@ def test_regrouping_realigns_occurrence_event_id(self):
         # No occurrence should be left pointing at a deleted/missing event.
         assert Occurrence.objects.filter(deployment=self.deployment, event__isnull=True).count() == 0
 
+    def _create_burst(self, start: datetime.datetime, n: int, interval_minutes: int = 5):
+        """Create ``n`` captures spaced ``interval_minutes`` apart starting at ``start``."""
+        import pathlib
+        import uuid
+
+        for i in range(n):
+            SourceImage.objects.create(
+                deployment=self.deployment,
+                timestamp=start + datetime.timedelta(minutes=i * interval_minutes),
+                path=pathlib.Path("test") / f"{uuid.uuid4().hex[:8]}_burst_{i}.jpg",
+            )
+
+    def test_cross_midnight_bursts_split_by_short_gap(self):
+        """
+        User-reported pattern: a "night" with a multi-hour off-window between
+        bursts that crosses midnight gets split into two separate Events at
+        the default 2 h gap, because the second burst's start_date is on the
+        next calendar day. Bumping ``max_time_gap`` past the off-window
+        merges both bursts into a single timestamp_group whose start_date
+        determines the group_by date, yielding one Event.
+        """
+        # Burst A: 22:00–22:25 on day N
+        self._create_burst(datetime.datetime(2023, 8, 5, 22, 0), n=6, interval_minutes=5)
+        # Burst B: 01:00–01:25 on day N+1 (off-window = 2 h 35 min, crosses midnight)
+        self._create_burst(datetime.datetime(2023, 8, 6, 1, 0), n=6, interval_minutes=5)
+
+        # Default 2 h gap → splits into two Events on different dates.
+        events_default = group_images_into_events(
+            deployment=self.deployment,
+            max_time_gap=datetime.timedelta(hours=2),
+        )
+        assert len(events_default) == 2, f"expected 2 Events at 2h gap, got {len(events_default)}"
+
+        # 4 h gap (off-window 2h35m < 4h) → single timestamp_group → one Event.
+        events_widened = group_images_into_events(
+            deployment=self.deployment,
+            max_time_gap=datetime.timedelta(hours=4),
+        )
+        assert len(events_widened) == 1, f"expected 1 Event at 4h gap, got {len(events_widened)}"
+
+    def test_same_date_bursts_merge_regardless_of_gap(self):
+        """
+        Inverse of the above: bursts that DON'T cross midnight collide on
+        ``group_by = start_date.date()`` even when the gap setting would
+        split them into separate timestamp_groups. This is the #904 caveat
+        — sub-day session-splits are masked by date-keyed Event reuse.
+        Documenting current behavior so a future fix (e.g. noon-to-noon
+        keying) has a regression target.
+        """
+        # Two bursts on the SAME calendar date with a 1 h 35 min off-window between.
+        self._create_burst(datetime.datetime(2023, 8, 5, 20, 0), n=6, interval_minutes=5)
+        self._create_burst(datetime.datetime(2023, 8, 5, 22, 0), n=6, interval_minutes=5)
+
+        # 1 h gap → splits into two timestamp_groups, but both start_date.date()
+        # is 2023-08-05 → get_or_create collides → ONE Event.
+        # group_images_into_events returns one entry per timestamp_group (with
+        # duplicates when groups reuse the same Event), so assert on the DB count.
+        group_images_into_events(
+            deployment=self.deployment,
+            max_time_gap=datetime.timedelta(hours=1),
+        )
+        db_event_count = Event.objects.filter(deployment=self.deployment).count()
+        assert db_event_count == 1, f"expected 1 Event due to date collision, got {db_event_count}"
+
+    def test_session_time_gap_seconds_is_used_when_no_explicit_gap(self):
+        """
+        When ``max_time_gap`` is not passed, ``group_images_into_events``
+        falls back to ``deployment.project.session_time_gap_seconds``. Verify
+        the project setting actually drives the split decision.
+        """
+        # Same cross-midnight burst pattern as the first test.
+        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)
+
+        # Project setting at 2 h (= 7200 s) → splits across midnight.
+        self.project.session_time_gap_seconds = 2 * 60 * 60
+        self.project.save()
+        # Bust the cached deployment.project relation so the function reads the new value.
+        self.deployment.refresh_from_db()
+        group_images_into_events(deployment=self.deployment)
+        count_2h = Event.objects.filter(deployment=self.deployment).count()
+        assert count_2h == 2, f"expected 2 Events at project setting 7200s, got {count_2h}"
+
+        # Project setting at 4 h (= 14400 s) → off-window 2h35m fits → single Event.
+        self.project.session_time_gap_seconds = 4 * 60 * 60
+        self.project.save()
+        self.deployment.refresh_from_db()
+        group_images_into_events(deployment=self.deployment)
+        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_is_idempotent_under_concurrent_calls(self):
+        """
+        ``group_images_into_events`` holds a per-deployment cache lock so
+        concurrent calls (admin, API action, autoregroup-on-save, sync-stage)
+        collapse to a single in-flight run. The second call should return an
+        empty list without creating any Events.
+        """
+        from django.core.cache import cache
+
+        lock_key = f"regroup_events:lock:deployment:{self.deployment.pk}"
+        # Make sure no stale lock from a prior test leaks in.
+        cache.delete(lock_key)
+
+        self._create_burst(datetime.datetime(2023, 8, 5, 22, 0), n=6, interval_minutes=5)
+        Event.objects.filter(deployment=self.deployment).delete()
+
+        # Pre-take the lock with a token we control.
+        cache.add(lock_key, "other-run-token", timeout=60)
+        try:
+            result = group_images_into_events(deployment=self.deployment)
+        finally:
+            cache.delete(lock_key)
+
+        assert result == [], "Locked run should return empty list"
+        assert Event.objects.filter(deployment=self.deployment).count() == 0, "Locked run should not create Events"
+
+        # Without the pre-taken lock, the next call should run normally and release the lock.
+        events = group_images_into_events(deployment=self.deployment)
+        assert len(events) >= 1, "Unlocked run should produce Events"
+        assert cache.get(lock_key) is None, "Lock should be released after the function returns"
+
+    def test_regroup_lock_release_does_not_clobber_a_newer_owner(self):
+        """
+        Token-based lock release: if our run takes longer than the lock TTL and
+        another caller acquires the same key with a fresh token, our ``finally``
+        block must not delete the other caller's lock.
+        """
+        from django.core.cache import cache
+
+        from ami.main.models import _regroup_lock
+
+        lock_key = f"regroup_events:lock:deployment:{self.deployment.pk}"
+        cache.delete(lock_key)
+
+        with _regroup_lock(self.deployment.pk) as acquired:
+            assert acquired, "Should acquire the lock on first try"
+            # Simulate the TTL expiring and a newer caller taking the lock.
+            cache.set(lock_key, "newer-owner-token", timeout=60)
+
+        # Our `finally` ran. The newer owner's lock should still be there.
+        assert cache.get(lock_key) == "newer-owner-token", "Expired-lock release must not clobber the newer owner"
+        cache.delete(lock_key)
+
     def test_pruning_empty_events(self):
         from ami.main.models import delete_empty_events
 
diff --git a/ami/tasks.py b/ami/tasks.py
index bc4400e47..040de7401 100644
--- a/ami/tasks.py
+++ b/ami/tasks.py
@@ -86,8 +86,15 @@ def populate_collection(collection_id: int) -> None:
     collection.populate_sample()
 
 
-# Task to group images into events
-@celery_app.task(soft_time_limit=one_hour, time_limit=one_hour + 60)
+# Task to group images into events. Kept as a thin Celery wrapper for
+# the Deployment.save() autoregroup path (which fires too often to spawn a Job).
+# The new RegroupEventsJob is the user-facing entry point — see ami/jobs/models.py.
+# Idempotency lives inside group_images_into_events() itself (see _regroup_lock),
+# so every caller — this task, the JobType, the admin bulk action, the autoregroup —
+# is protected without each having to wrap the call in a lock.
+# Regroup finishes in seconds; short limits keep the lock TTL short.
+# See REGROUP_LOCK_TTL_SECONDS in ami.main.models for the matching value.
+@celery_app.task(soft_time_limit=10 * 60, time_limit=11 * 60)
 def regroup_events(deployment_id: int) -> None:
     from ami.main.models import Deployment, group_images_into_events
 
@@ -98,7 +105,13 @@ def regroup_events(deployment_id: int) -> None:
         return
     logger.info(f"Grouping captures for {deployment}")
     events = group_images_into_events(deployment)
-    logger.info(f"{deployment} now has {len(events)} events")
+    if events:
+        logger.info(f"{deployment} now has {len(events)} events")
+    else:
+        # Empty return = lock-miss short-circuit (see _regroup_lock); the WARNING
+        # from group_images_into_events above explains why. Don't log "now has 0
+        # events" — it's misleading: nothing was reset.
+        logger.info(f"{deployment} regroup skipped — see warning above")
 
 
 @celery_app.task(soft_time_limit=one_hour, time_limit=one_hour + 60)
diff --git a/ami/users/roles.py b/ami/users/roles.py
index dfb2e4ca6..5a5f9f6c8 100644
--- a/ami/users/roles.py
+++ b/ami/users/roles.py
@@ -141,6 +141,7 @@ class MLDataManager(Role):
         Project.Permissions.RUN_ML_JOB,
         Project.Permissions.RUN_POPULATE_CAPTURES_COLLECTION_JOB,
         Project.Permissions.RUN_DATA_STORAGE_SYNC_JOB,
+        Project.Permissions.RUN_REGROUP_EVENTS_JOB,
         Project.Permissions.RUN_DATA_EXPORT_JOB,
         Project.Permissions.DELETE_OCCURRENCES,
         Project.Permissions.CREATE_PROJECT_PIPELINE_CONFIG,
@@ -181,6 +182,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,