diff --git a/api_app/models.py b/api_app/models.py
index 95ecc65668..341debd36e 100644
--- a/api_app/models.py
+++ b/api_app/models.py
@@ -457,9 +457,27 @@ def get_root(self):
         try:
             return super().get_root()
         except self.MultipleObjectsReturned:
-            # django treebeard is not thread safe
-            # this is not a really valid solution, but it will work for now
-            return self.objects.filter(path=self.path[0 : self.steplen]).first()  # noqa
+            # django-treebeard is not thread safe - this indicates a data
+            # integrity issue. We intentionally avoid select_for_update() to
+            # prevent potential database deadlocks in high-concurrency
+            # environments (e.g., multiple Celery workers).
+            # Using order_by('pk').first() ensures all concurrent requests
+            # get the same deterministic root node, even if multiple roots
+            # exist due to a race condition.
+            # Note: The root cause is in django-treebeard's tree modification
+            # operations, not in this read operation.
+            root_node = (
+                type(self)
+                .objects.filter(path=self.path[: self.steplen])
+                .order_by("pk")
+                .first()
+            )
+            logger.warning(
+                f"Tree Integrity Error: Multiple roots found for Job {self.pk} "
+                f"(path: {self.path}). Returning deterministic root "
+                f"(PK: {root_node.pk if root_node else 'None'}) as fallback."
+            )
+            return root_node
 
     @cached_property
     def is_sample(self) -> bool:
diff --git a/tests/api_app/test_models.py b/tests/api_app/test_models.py
index 06208f9ed1..9e74387341 100644
--- a/tests/api_app/test_models.py
+++ b/tests/api_app/test_models.py
@@ -547,3 +547,123 @@ def test_pivots_to_execute(self):
         self.assertCountEqual(
             j1.pivots_to_execute.filter(name="test").values_list("pk", flat=True), []
         )
+
+    def test_get_root_returns_self_when_is_root(self):
+        """Test that get_root() returns self when the job is already a root node."""
+        an = Analyzable.objects.create(
+            name="test.com",
+            classification=Classification.DOMAIN,
+        )
+        root_job = Job.objects.create(
+            user=self.user,
+            analyzable=an,
+            status=Job.STATUSES.REPORTED_WITHOUT_FAILS,
+        )
+        # A newly created job should be a root
+        self.assertTrue(root_job.is_root())
+        self.assertEqual(root_job.get_root(), root_job)
+        root_job.delete()
+        an.delete()
+
+    def test_get_root_returns_parent_for_child_job(self):
+        """Test that get_root() returns the root job for a child job."""
+        an = Analyzable.objects.create(
+            name="test.com",
+            classification=Classification.DOMAIN,
+        )
+        root_job = Job.add_root(
+            user=self.user,
+            analyzable=an,
+            status=Job.STATUSES.REPORTED_WITHOUT_FAILS,
+        )
+        child_job = root_job.add_child(
+            user=self.user,
+            analyzable=an,
+            status=Job.STATUSES.REPORTED_WITHOUT_FAILS,
+        )
+        # Child job should return the root job
+        self.assertFalse(child_job.is_root())
+        self.assertEqual(child_job.get_root().pk, root_job.pk)
+        child_job.delete()
+        root_job.delete()
+        an.delete()
+
+    def test_get_root_deterministic_ordering(self):
+        """Test that get_root() returns deterministic results using order_by('pk')."""
+        an = Analyzable.objects.create(
+            name="test.com",
+            classification=Classification.DOMAIN,
+        )
+        root_job = Job.add_root(
+            user=self.user,
+            analyzable=an,
+            status=Job.STATUSES.REPORTED_WITHOUT_FAILS,
+        )
+        # Call get_root multiple times and verify consistent results
+        results = [root_job.get_root().pk for _ in range(10)]
+        self.assertEqual(
+            len(set(results)), 1, "get_root() should return consistent results"
+        )
+        root_job.delete()
+        an.delete()
+
+    def test_get_root_handles_multiple_roots_deterministically(self):
+        """
+        Test that get_root() handles MultipleObjectsReturned exception
+        by returning a deterministic result based on PK ordering.
+
+        This simulates the race condition that can occur with django-treebeard
+        under high concurrency. We use mocking because the path field has a
+        UNIQUE constraint in the database, preventing real duplicates.
+        """
+        from unittest.mock import patch
+
+        an = Analyzable.objects.create(
+            name="test.com",
+            classification=Classification.DOMAIN,
+        )
+        # Create a root job and child job
+        root_job = Job.add_root(
+            user=self.user,
+            analyzable=an,
+            status=Job.STATUSES.REPORTED_WITHOUT_FAILS,
+        )
+        child_job = root_job.add_child(
+            user=self.user,
+            analyzable=an,
+            status=Job.STATUSES.REPORTED_WITHOUT_FAILS,
+        )
+
+        # Verify child_job is not a root (needed for the test to work)
+        self.assertFalse(child_job.is_root())
+
+        # Import MP_Node to patch its get_root method
+        from treebeard.mp_tree import MP_Node
+
+        # Patch treebeard's MP_Node.get_root to raise MultipleObjectsReturned
+        # and also patch the logger to verify it was called
+        with (
+            patch.object(
+                MP_Node,
+                "get_root",
+                side_effect=Job.MultipleObjectsReturned("Multiple roots found"),
+            ),
+            patch("api_app.models.logger") as mock_logger,
+        ):
+            result = child_job.get_root()
+
+        # Verify we got a result (the fallback query should work)
+        self.assertIsNotNone(result)
+        # The fallback query finds root_job (the only actual root)
+        self.assertEqual(result.pk, root_job.pk)
+
+        # Verify warning was logged (using mock to avoid CI logging disable issues)
+        mock_logger.warning.assert_called_once()
+        call_args = mock_logger.warning.call_args[0][0]
+        self.assertIn("Tree Integrity Error", call_args)
+        self.assertIn("Multiple roots found", call_args)
+
+        # Cleanup
+        child_job.delete()
+        root_job.delete()
+        an.delete()