diff --git a/ami/main/api/serializers.py b/ami/main/api/serializers.py
index 6c899c2f7..ca502da38 100644
--- a/ami/main/api/serializers.py
+++ b/ami/main/api/serializers.py
@@ -1,3 +1,4 @@
+import collections
 import datetime
 
 from django.db.models import QuerySet
@@ -800,6 +801,80 @@ class Meta:
         ]
 
 
+#: Upper bound on a single bulk identification request.
+#: Every identification in a batch is written inside the request's transaction, which
+#: holds row locks on each occurrence until the request finishes, so an unbounded batch
+#: would block other people identifying the same occurrences for as long as it ran. The
+#: identification interface can only select occurrences on the page being displayed, so
+#: real batches are far smaller than this; the cap bounds the worst case rather than
+#: shaping the interface.
+MAX_BULK_IDENTIFICATIONS = 200
+
+
+class BulkIdentificationItemSerializer(serializers.Serializer):
+    """
+    One identification within a bulk request.
+
+    Related objects are declared as plain integers rather than
+    `PrimaryKeyRelatedField` so that the view can resolve the whole batch in a
+    fixed number of queries. `PrimaryKeyRelatedField` issues one query per field
+    per item, which would make validation alone scale with the size of the batch.
+
+    `withdrawn` is deliberately absent: the model maintains it, and letting a
+    client set it would break the "one active identification per user per
+    occurrence" invariant.
+    """
+
+    occurrence_id = serializers.IntegerField()
+    taxon_id = serializers.IntegerField()
+    comment = serializers.CharField(required=False, allow_blank=True, default="")
+    agreed_with_identification_id = serializers.IntegerField(required=False, allow_null=True, default=None)
+    agreed_with_prediction_id = serializers.IntegerField(required=False, allow_null=True, default=None)
+
+
+class BulkIdentificationRequestSerializer(serializers.Serializer):
+    """Validates the shape of a bulk request, before any occurrence is looked up."""
+
+    identifications = BulkIdentificationItemSerializer(many=True, allow_empty=False)
+
+    def validate_identifications(self, value: list[dict]) -> list[dict]:
+        if len(value) > MAX_BULK_IDENTIFICATIONS:
+            raise serializers.ValidationError(
+                f"A single request may contain at most {MAX_BULK_IDENTIFICATIONS} identifications, "
+                f"got {len(value)}."
+            )
+
+        counts = collections.Counter(item["occurrence_id"] for item in value)
+        duplicates = sorted(pk for pk, count in counts.items() if count > 1)
+        if duplicates:
+            # Two identifications for one occurrence in one batch have no defined
+            # winner: the outcome would depend on insert ordering rather than on
+            # anything the client asked for.
+            raise serializers.ValidationError(
+                f"Each occurrence may appear only once per request. Repeated occurrence IDs: {duplicates}."
+            )
+
+        return value
+
+
+class BulkIdentificationResultSerializer(serializers.Serializer):
+    """The outcome of a single submitted item, matched to the request by `index`."""
+
+    index = serializers.IntegerField()
+    occurrence_id = serializers.IntegerField()
+    status = serializers.ChoiceField(choices=["created", "error"])
+    id = serializers.IntegerField(required=False)
+    errors = serializers.DictField(required=False)
+
+
+class BulkIdentificationResponseSerializer(serializers.Serializer):
+    """Per-item outcomes for a bulk request, in the order the items were submitted."""
+
+    created_count = serializers.IntegerField()
+    error_count = serializers.IntegerField()
+    results = BulkIdentificationResultSerializer(many=True)
+
+
 class TaxonDetectionsSerializer(DefaultSerializer):
     class Meta:
         model = Detection
diff --git a/ami/main/api/views.py b/ami/main/api/views.py
index 591a4a000..9fecbd518 100644
--- a/ami/main/api/views.py
+++ b/ami/main/api/views.py
@@ -6,7 +6,7 @@
 from django.contrib.postgres.search import TrigramSimilarity
 from django.core import exceptions
 from django.core.files.storage import default_storage
-from django.db import models
+from django.db import DatabaseError, models, transaction
 from django.db.models import OuterRef, Prefetch, Q, Subquery
 from django.db.models.query import QuerySet
 from django.forms import BooleanField, CharField, IntegerField
@@ -62,6 +62,8 @@
     update_detection_counts,
 )
 from .serializers import (
+    BulkIdentificationRequestSerializer,
+    BulkIdentificationResponseSerializer,
     ClassificationListSerializer,
     ClassificationSerializer,
     ClassificationWithTaxaSerializer,
@@ -2237,6 +2239,223 @@ def perform_create(self, serializer):
 
         serializer.save(user=self.request.user)
 
+    @extend_schema(
+        request=BulkIdentificationRequestSerializer,
+        responses={200: BulkIdentificationResponseSerializer},
+    )
+    @action(detail=False, methods=["post"])
+    def bulk(self, request):
+        """
+        Create many identifications in one request.
+
+        Applying determinations to a page of occurrences otherwise costs one HTTP
+        request per occurrence. Each item is saved independently, so a single bad
+        item (an occurrence deleted since the page was loaded, say) reports an
+        error against its own index while the rest of the batch still succeeds.
+        The response lists an outcome per submitted item, in request order.
+        """
+        request_serializer = BulkIdentificationRequestSerializer(data=request.data)
+        request_serializer.is_valid(raise_exception=True)
+        items = request_serializer.validated_data["identifications"]
+
+        occurrences = self._resolve_occurrences(items)
+        self._authorize_batch(request, occurrences)
+        taxa = self._resolve_taxa(items)
+        agreed_identifications, agreed_predictions = self._resolve_agreement_targets(items)
+
+        results = []
+        created_count = 0
+        for index, item in enumerate(items):
+            errors = self._validate_item(item, occurrences, taxa, agreed_identifications, agreed_predictions)
+            if errors:
+                results.append(
+                    {
+                        "index": index,
+                        "occurrence_id": item["occurrence_id"],
+                        "status": "error",
+                        "errors": errors,
+                    }
+                )
+                continue
+
+            identification = Identification(
+                occurrence=occurrences[item["occurrence_id"]],
+                taxon=taxa[item["taxon_id"]],
+                user=request.user,
+                comment=item["comment"],
+                agreed_with_identification=agreed_identifications.get(item["agreed_with_identification_id"]),
+                agreed_with_prediction=agreed_predictions.get(item["agreed_with_prediction_id"]),
+            )
+            try:
+                # ATOMIC_REQUESTS wraps the whole request in a transaction, so this
+                # is a savepoint rather than a separate transaction. Rolling back to
+                # it leaves the connection usable, which is what lets the rest of the
+                # batch continue after one item fails. Without catching the error
+                # here, a single failure would abort the request and discard every
+                # identification already made in it.
+                #
+                # save() also withdraws the user's earlier identification on this
+                # occurrence and recomputes the occurrence's determination.
+                with transaction.atomic():
+                    identification.save()
+            except (DatabaseError, exceptions.ObjectDoesNotExist) as error:
+                # The occurrence can be deleted between resolving the batch and
+                # writing it, which surfaces as an integrity error or a missing row
+                # part-way through save().
+                logger.warning(
+                    f"Bulk identification of occurrence {item['occurrence_id']} failed and was skipped: {error}"
+                )
+                results.append(
+                    {
+                        "index": index,
+                        "occurrence_id": item["occurrence_id"],
+                        "status": "error",
+                        "errors": {
+                            "occurrence_id": ["This occurrence could not be identified. It may have been deleted."]
+                        },
+                    }
+                )
+                continue
+
+            created_count += 1
+            results.append(
+                {
+                    "index": index,
+                    "occurrence_id": item["occurrence_id"],
+                    "status": "created",
+                    "id": identification.pk,
+                }
+            )
+
+        return Response(
+            {
+                "created_count": created_count,
+                "error_count": len(items) - created_count,
+                "results": results,
+            }
+        )
+
+    def _resolve_occurrences(self, items: list[dict]) -> dict[int, Occurrence]:
+        """Fetch every occurrence in the batch in one query, with its project loaded."""
+        occurrence_ids = {item["occurrence_id"] for item in items}
+        # select_related("project") lets the permission probe and the single-project
+        # check read the project without a query per occurrence.
+        return {
+            occurrence.pk: occurrence
+            for occurrence in Occurrence.objects.filter(pk__in=occurrence_ids).select_related("project")
+        }
+
+    def _authorize_batch(self, request, occurrences: dict[int, Occurrence]) -> None:
+        """
+        Authorize the batch against the single project it belongs to.
+
+        A `detail=False` action never passes through `has_object_permission`, so the
+        check has to happen here. Permission to create an identification is granted
+        per project, so one check per batch is equivalent to one check per
+        occurrence — see the note on `Identification.check_permission`.
+        """
+        if not occurrences:
+            # Nothing resolved, so there is no project to authorize against and
+            # nothing to write. Every item is reported as not found instead, which
+            # keeps a batch of one missing occurrence answering the same way as a
+            # batch where only some are missing.
+            return
+
+        projects = {occurrence.project for occurrence in occurrences.values()}
+        if len(projects) > 1:
+            raise api_exceptions.ValidationError(
+                {"identifications": ["All occurrences in a request must belong to the same project."]}
+            )
+
+        probe = Identification(occurrence=next(iter(occurrences.values())))
+        if not probe.check_permission(request.user, "create"):
+            # "create", not the name of this action: the action name is not in the
+            # model's CRUD permission map and would silently fall through to a
+            # permission that no role grants.
+            raise PermissionDenied("You do not have permission to identify occurrences in this project.")
+
+    def _resolve_taxa(self, items: list[dict]) -> dict[int, Taxon]:
+        """Fetch every taxon in the batch in one query."""
+        taxon_ids = {item["taxon_id"] for item in items}
+        return {taxon.pk: taxon for taxon in Taxon.objects.filter(pk__in=taxon_ids)}
+
+    def _resolve_agreement_targets(
+        self, items: list[dict]
+    ) -> tuple[dict[int, Identification], dict[int, Classification]]:
+        """
+        Fetch the identifications and predictions that items claim to agree with.
+
+        Both are recorded only as provenance, but they still have to be real and
+        belong to the occurrence being identified.
+        """
+        identification_ids = {
+            item["agreed_with_identification_id"] for item in items if item["agreed_with_identification_id"]
+        }
+        prediction_ids = {item["agreed_with_prediction_id"] for item in items if item["agreed_with_prediction_id"]}
+
+        agreed_identifications = {}
+        if identification_ids:
+            agreed_identifications = {
+                identification.pk: identification
+                for identification in Identification.objects.filter(pk__in=identification_ids)
+            }
+
+        agreed_predictions = {}
+        if prediction_ids:
+            agreed_predictions = {
+                classification.pk: classification
+                for classification in Classification.objects.filter(pk__in=prediction_ids).select_related("detection")
+            }
+
+        return agreed_identifications, agreed_predictions
+
+    def _validate_item(
+        self,
+        item: dict,
+        occurrences: dict[int, Occurrence],
+        taxa: dict[int, Taxon],
+        agreed_identifications: dict[int, Identification],
+        agreed_predictions: dict[int, Classification],
+    ) -> dict[str, list[str]]:
+        """Check one item against the already-fetched batch. Returns field errors, empty when valid."""
+        errors: dict[str, list[str]] = {}
+
+        occurrence = occurrences.get(item["occurrence_id"])
+        if occurrence is None:
+            errors["occurrence_id"] = [f"Occurrence {item['occurrence_id']} does not exist."]
+
+        if item["taxon_id"] not in taxa:
+            errors["taxon_id"] = [f"Taxon {item['taxon_id']} does not exist."]
+
+        agreed_identification_id = item["agreed_with_identification_id"]
+        if agreed_identification_id:
+            agreed = agreed_identifications.get(agreed_identification_id)
+            if agreed is None:
+                errors["agreed_with_identification_id"] = [
+                    f"Identification {agreed_identification_id} does not exist."
+                ]
+            elif occurrence is not None and agreed.occurrence_id != occurrence.pk:
+                errors["agreed_with_identification_id"] = [
+                    "An identification can only agree with another identification of the same occurrence."
+                ]
+
+        agreed_prediction_id = item["agreed_with_prediction_id"]
+        if agreed_prediction_id:
+            prediction = agreed_predictions.get(agreed_prediction_id)
+            if prediction is None:
+                errors["agreed_with_prediction_id"] = [f"Classification {agreed_prediction_id} does not exist."]
+            elif occurrence is not None and (
+                # A classification keeps its row when its detection is deleted, so the
+                # link back to an occurrence can be missing entirely.
+                prediction.detection is None
+                or prediction.detection.occurrence_id != occurrence.pk
+            ):
+                errors["agreed_with_prediction_id"] = [
+                    "An identification can only agree with a prediction of the same occurrence."
+                ]
+
+        return errors
+
 
 class SiteViewSet(DefaultViewSet, ProjectMixin):
     """
diff --git a/ami/main/models.py b/ami/main/models.py
index 3662b4107..22c1197e0 100644
--- a/ami/main/models.py
+++ b/ami/main/models.py
@@ -2912,7 +2912,15 @@ def delete(self, *args, **kwargs):
         update_occurrence_determination(self.occurrence, current_determination=self.taxon)
 
     def check_permission(self, user: AbstractUser | AnonymousUser, action: str) -> bool:
-        """Custom permission check logic for Identification model."""
+        """
+        Custom permission check logic for Identification model.
+
+        The "create" branch depends only on the occurrence's project, never on the
+        occurrence itself. The bulk identification endpoint relies on that: it checks
+        "create" once for the whole batch instead of once per occurrence. Adding
+        per-occurrence logic to the "create" path means that endpoint has to check
+        each occurrence individually.
+        """
         import ami.users.roles as roles
 
         project = self.get_project()
diff --git a/ami/main/test_bulk_identifications.py b/ami/main/test_bulk_identifications.py
new file mode 100644
index 000000000..f8ecb0b36
--- /dev/null
+++ b/ami/main/test_bulk_identifications.py
@@ -0,0 +1,578 @@
+"""
+Tests for the bulk identifications endpoint.
+
+The endpoint lets an identifier apply determinations to many occurrences in one
+request. The tests here pin the parts of that contract that are easy to break
+without noticing: per-occurrence permission enforcement, the withdraw-previous
+and determination-recompute side effects that live in ``Identification.save()``,
+and the fact that validation cost must not grow with the size of the batch.
+"""
+
+import logging
+from unittest import mock
+
+from django.db import IntegrityError, connection
+from django.test.utils import CaptureQueriesContext
+from rest_framework import status
+from rest_framework.test import APITestCase
+
+from ami.main.api.serializers import MAX_BULK_IDENTIFICATIONS
+from ami.main.models import Identification, Occurrence, Taxon
+from ami.tests.fixtures.main import create_captures, create_occurrences, create_taxa, setup_test_project
+from ami.users.models import User
+from ami.users.roles import BasicMember, Identifier, ProjectManager
+
+logger = logging.getLogger(__name__)
+
+ENDPOINT = "/api/v2/identifications/bulk/"
+
+
+class BulkIdentificationTestCase(APITestCase):
+    """Shared fixture: one project with several occurrences and a user per role."""
+
+    def setUp(self) -> None:
+        self.project, self.deployment = setup_test_project(reuse=False)
+        create_taxa(project=self.project)
+        create_captures(deployment=self.deployment)
+        create_occurrences(deployment=self.deployment, num=4)
+
+        self.identifier = User.objects.create_user(email="identifier@insectai.org")  # type: ignore[attr-defined]
+        self.basic_member = User.objects.create_user(email="basic@insectai.org")  # type: ignore[attr-defined]
+        self.non_member = User.objects.create_user(email="stranger@insectai.org")  # type: ignore[attr-defined]
+        self.superuser = User.objects.create_user(  # type: ignore[attr-defined]
+            email="super@insectai.org", is_staff=True, is_superuser=True
+        )
+        Identifier.assign_user(self.identifier, self.project)
+        BasicMember.assign_user(self.basic_member, self.project)
+        ProjectManager.assign_user(self.superuser, self.project)
+
+        self.occurrences = list(Occurrence.objects.filter(project=self.project).exclude(determination=None))
+        assert len(self.occurrences) >= 4, "Fixture must provide enough occurrences to catch per-row query growth"
+
+        self.taxon = Taxon.objects.exclude(pk=self.occurrences[0].determination_id).first()
+        assert self.taxon is not None
+
+        return super().setUp()
+
+    def post_bulk(self, items: list[dict], user: User | None = None):
+        if user is not None:
+            self.client.force_authenticate(user=user)
+        return self.client.post(ENDPOINT, {"identifications": items}, format="json")
+
+    def item(self, occurrence: Occurrence, taxon: Taxon | None = None, **extra) -> dict:
+        return {"occurrence_id": occurrence.pk, "taxon_id": (taxon or self.taxon).pk, **extra}
+
+
+class TestBulkIdentificationSuccess(BulkIdentificationTestCase):
+    def test_creates_one_identification_per_item_and_updates_determinations(self):
+        """The happy path: every item becomes an Identification and moves its occurrence's determination."""
+        targets = self.occurrences[:3]
+        response = self.post_bulk([self.item(occurrence) for occurrence in targets], user=self.identifier)
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK, response.content)
+        body = response.json()
+        self.assertEqual(body["created_count"], 3)
+        self.assertEqual(body["error_count"], 0)
+        self.assertEqual([result["status"] for result in body["results"]], ["created"] * 3)
+
+        for occurrence in targets:
+            occurrence.refresh_from_db()
+            self.assertEqual(occurrence.determination, self.taxon)
+            self.assertEqual(
+                Identification.objects.filter(occurrence=occurrence, user=self.identifier, withdrawn=False).count(),
+                1,
+            )
+
+    def test_accepts_ids_sent_as_strings(self):
+        """
+        The frontend sends occurrence and taxon IDs as JSON strings.
+
+        The request body is built from the identification form, where IDs are
+        strings, so the endpoint has to accept "123" and not only 123. This pins
+        the frontend-to-backend contract; a stricter integer-only field would 400
+        every real bulk identification.
+        """
+        occurrence = self.occurrences[0]
+        response = self.post_bulk(
+            [{"occurrence_id": str(occurrence.pk), "taxon_id": str(self.taxon.pk)}],
+            user=self.identifier,
+        )
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK, response.content)
+        self.assertEqual(response.json()["created_count"], 1)
+
+    def test_results_are_returned_in_request_order(self):
+        """Clients match results to submitted items by index, so order and index must be stable."""
+        targets = self.occurrences[:3]
+        response = self.post_bulk([self.item(occurrence) for occurrence in targets], user=self.identifier)
+
+        body = response.json()
+        self.assertEqual([result["index"] for result in body["results"]], [0, 1, 2])
+        self.assertEqual(
+            [result["occurrence_id"] for result in body["results"]],
+            [occurrence.pk for occurrence in targets],
+        )
+
+    def test_comment_is_saved(self):
+        response = self.post_bulk(
+            [self.item(self.occurrences[0], comment="Wing pattern matches")], user=self.identifier
+        )
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK, response.content)
+        identification = Identification.objects.get(pk=response.json()["results"][0]["id"])
+        self.assertEqual(identification.comment, "Wing pattern matches")
+
+    def test_withdraws_previous_identification_by_the_same_user(self):
+        """
+        A user has one active identification per occurrence.
+
+        ``Identification.save()`` withdraws the user's earlier identifications on that
+        occurrence. A test using only fresh occurrences passes whether or not the bulk
+        path preserves that, so this pins it with a pre-existing identification.
+        """
+        occurrence = self.occurrences[0]
+        previous = Identification.objects.create(occurrence=occurrence, taxon=self.taxon, user=self.identifier)
+        self.assertFalse(previous.withdrawn)
+
+        other_taxon = Taxon.objects.exclude(pk__in=[self.taxon.pk]).first()
+        assert other_taxon is not None
+        response = self.post_bulk([self.item(occurrence, taxon=other_taxon)], user=self.identifier)
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK, response.content)
+        previous.refresh_from_db()
+        self.assertTrue(previous.withdrawn, "The user's earlier identification must be withdrawn")
+        occurrence.refresh_from_db()
+        self.assertEqual(occurrence.determination, other_taxon)
+
+    def test_does_not_withdraw_identifications_by_other_users(self):
+        """Withdrawing is scoped to the submitting user; another identifier's opinion must survive."""
+        occurrence = self.occurrences[0]
+        other_user_id = Identification.objects.create(occurrence=occurrence, taxon=self.taxon, user=self.superuser)
+
+        self.post_bulk([self.item(occurrence)], user=self.identifier)
+
+        other_user_id.refresh_from_db()
+        self.assertFalse(other_user_id.withdrawn)
+
+    def test_agreeing_with_the_current_determination_raises_the_score_to_the_human_score(self):
+        """
+        Agreeing with an occurrence's existing determination keeps the taxon and lifts
+        the score to the human identification's score of 1.0.
+
+        This mirrors what a single POST to /identifications/ already does, which is the
+        behaviour that matters: the bulk endpoint is a faster way to do the same thing,
+        not a different thing. `determination_score` feeds the project's score-threshold
+        filters, so a bulk path that left the machine score in place would quietly change
+        which occurrences appear in a filtered list.
+        """
+        occurrence = self.occurrences[0]
+        original_taxon = occurrence.determination
+        self.assertEqual(occurrence.determination_score, 0.9)
+
+        response = self.post_bulk(
+            [self.item(occurrence, taxon=original_taxon, agreed_with_prediction_id=occurrence.best_prediction.pk)],
+            user=self.identifier,
+        )
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK, response.content)
+        occurrence.refresh_from_db()
+        self.assertEqual(occurrence.determination, original_taxon)
+        self.assertEqual(occurrence.determination_score, 1.0)
+
+    def test_bulk_and_single_post_leave_an_occurrence_in_the_same_state(self):
+        """
+        The bulk endpoint must be indistinguishable from the single-item endpoint.
+
+        Anything the bulk path reimplements or skips — withdrawing the user's previous
+        identification, recomputing the determination, the resulting score — shows up
+        here as a divergence, without this test having to name each rule.
+        """
+        via_single, via_bulk = self.occurrences[0], self.occurrences[1]
+
+        # Seed both with an earlier identification by the same user, so that
+        # withdraw-previous is part of what the two paths have to agree on.
+        other_taxon = Taxon.objects.exclude(pk=self.taxon.pk).first()
+        assert other_taxon is not None
+        for occurrence in (via_single, via_bulk):
+            Identification.objects.create(occurrence=occurrence, taxon=other_taxon, user=self.identifier)
+
+        self.client.force_authenticate(user=self.identifier)
+        single_response = self.client.post(
+            "/api/v2/identifications/",
+            {"occurrence_id": via_single.pk, "taxon_id": self.taxon.pk, "comment": "same"},
+            format="json",
+        )
+        self.assertEqual(single_response.status_code, status.HTTP_201_CREATED, single_response.content)
+
+        bulk_response = self.post_bulk([self.item(via_bulk, comment="same")], user=self.identifier)
+        self.assertEqual(bulk_response.status_code, status.HTTP_200_OK, bulk_response.content)
+
+        via_single.refresh_from_db()
+        via_bulk.refresh_from_db()
+        self.assertEqual(via_bulk.determination_id, via_single.determination_id)
+        self.assertEqual(via_bulk.determination_score, via_single.determination_score)
+
+        def identification_state(occurrence):
+            return sorted(
+                Identification.objects.filter(occurrence=occurrence).values_list("taxon_id", "withdrawn", "comment")
+            )
+
+        self.assertEqual(identification_state(via_bulk), identification_state(via_single))
+
+    def test_agreed_with_prediction_is_recorded(self):
+        """The agree provenance FK is stored so exports can report what the user agreed with."""
+        occurrence = self.occurrences[0]
+        prediction = occurrence.best_prediction
+        response = self.post_bulk(
+            [self.item(occurrence, taxon=occurrence.determination, agreed_with_prediction_id=prediction.pk)],
+            user=self.identifier,
+        )
+
+        identification = Identification.objects.get(pk=response.json()["results"][0]["id"])
+        self.assertEqual(identification.agreed_with_prediction_id, prediction.pk)
+
+
+class TestBulkIdentificationPartialFailure(BulkIdentificationTestCase):
+    def test_valid_items_are_saved_when_one_item_fails(self):
+        """
+        One bad item must not discard the rest of the batch.
+
+        Mass identification is the point of this endpoint; failing 49 good rows because
+        a 50th occurrence was deleted mid-session would be worse than the N-request
+        version it replaces.
+        """
+        items = [
+            self.item(self.occurrences[0]),
+            {"occurrence_id": 9_999_999, "taxon_id": self.taxon.pk},
+            self.item(self.occurrences[1]),
+        ]
+        response = self.post_bulk(items, user=self.identifier)
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK, response.content)
+        body = response.json()
+        self.assertEqual(body["created_count"], 2)
+        self.assertEqual(body["error_count"], 1)
+        self.assertEqual([result["status"] for result in body["results"]], ["created", "error", "created"])
+        self.assertIn("occurrence_id", body["results"][1]["errors"])
+
+        for occurrence in self.occurrences[:2]:
+            occurrence.refresh_from_db()
+            self.assertEqual(occurrence.determination, self.taxon)
+
+    def test_unknown_taxon_is_reported_per_item(self):
+        items = [self.item(self.occurrences[0]), {"occurrence_id": self.occurrences[1].pk, "taxon_id": 9_999_999}]
+        response = self.post_bulk(items, user=self.identifier)
+
+        body = response.json()
+        self.assertEqual(body["created_count"], 1)
+        self.assertIn("taxon_id", body["results"][1]["errors"])
+
+    def test_agreed_with_identification_from_another_occurrence_is_rejected(self):
+        """Agree provenance must point at the occurrence being identified, not an unrelated one."""
+        foreign = Identification.objects.create(occurrence=self.occurrences[1], taxon=self.taxon, user=self.superuser)
+        response = self.post_bulk(
+            [self.item(self.occurrences[0], agreed_with_identification_id=foreign.pk)],
+            user=self.identifier,
+        )
+
+        body = response.json()
+        self.assertEqual(body["created_count"], 0)
+        self.assertIn("agreed_with_identification_id", body["results"][0]["errors"])
+
+    def test_a_failed_item_does_not_roll_back_successful_items(self):
+        """A rejected item must not undo the identifications already made in the batch."""
+        items = [self.item(self.occurrences[0]), {"occurrence_id": self.occurrences[1].pk, "taxon_id": 9_999_999}]
+        self.post_bulk(items, user=self.identifier)
+
+        self.assertTrue(Identification.objects.filter(occurrence=self.occurrences[0]).exists())
+        self.assertFalse(Identification.objects.filter(occurrence=self.occurrences[1]).exists())
+
+    def test_a_database_failure_on_one_item_is_reported_without_losing_the_others(self):
+        """
+        A write that fails inside save() costs that item only.
+
+        The request runs in a single transaction (ATOMIC_REQUESTS), so each item is
+        saved inside a savepoint and its failure is caught. Without that, the first
+        failure would abort the transaction, discard every identification already
+        made in the request, and return a 500 instead of a per-item error. This is
+        the case that made the endpoint's partial-success promise real, so it is
+        pinned by forcing a failure rather than by trusting the arrangement.
+        """
+        failing_occurrence_id = self.occurrences[1].pk
+        original_save = Identification.save
+
+        def save_but_fail_for_one(identification_self, *args, **kwargs):
+            if identification_self.occurrence_id == failing_occurrence_id:
+                raise IntegrityError("simulated conflict while saving")
+            return original_save(identification_self, *args, **kwargs)
+
+        items = [self.item(occurrence) for occurrence in self.occurrences[:3]]
+        with mock.patch.object(Identification, "save", save_but_fail_for_one):
+            response = self.post_bulk(items, user=self.identifier)
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK, response.content)
+        body = response.json()
+        self.assertEqual(body["created_count"], 2)
+        self.assertEqual(body["error_count"], 1)
+        self.assertEqual([result["status"] for result in body["results"]], ["created", "error", "created"])
+
+        # The surviving items are really committed, not merely reported as created.
+        self.assertTrue(Identification.objects.filter(occurrence=self.occurrences[0]).exists())
+        self.assertFalse(Identification.objects.filter(occurrence=self.occurrences[1]).exists())
+        self.assertTrue(Identification.objects.filter(occurrence=self.occurrences[2]).exists())
+
+    def test_every_occurrence_missing_is_reported_per_item(self):
+        """
+        A batch where nothing resolves answers like any other batch of failures.
+
+        A batch of one deleted occurrence and a batch where only some are deleted are
+        the same kind of failure, so they must not return different shapes.
+        """
+        response = self.post_bulk([{"occurrence_id": 9_999_998, "taxon_id": self.taxon.pk}], user=self.identifier)
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK, response.content)
+        body = response.json()
+        self.assertEqual(body["created_count"], 0)
+        self.assertEqual(body["error_count"], 1)
+        self.assertIn("occurrence_id", body["results"][0]["errors"])
+
+    def test_unknown_agreement_targets_are_reported_per_item(self):
+        response = self.post_bulk(
+            [
+                self.item(self.occurrences[0], agreed_with_identification_id=9_999_999),
+                self.item(self.occurrences[1], agreed_with_prediction_id=9_999_999),
+            ],
+            user=self.identifier,
+        )
+
+        body = response.json()
+        self.assertEqual(body["created_count"], 0)
+        self.assertIn("agreed_with_identification_id", body["results"][0]["errors"])
+        self.assertIn("agreed_with_prediction_id", body["results"][1]["errors"])
+
+    def test_agreed_with_prediction_from_another_occurrence_is_rejected(self):
+        foreign_prediction = self.occurrences[1].best_prediction
+        response = self.post_bulk(
+            [self.item(self.occurrences[0], agreed_with_prediction_id=foreign_prediction.pk)],
+            user=self.identifier,
+        )
+
+        body = response.json()
+        self.assertEqual(body["created_count"], 0)
+        self.assertIn("agreed_with_prediction_id", body["results"][0]["errors"])
+
+    def test_a_missing_occurrence_does_not_produce_a_spurious_agreement_error(self):
+        """
+        With no occurrence to compare against, the agreement target cannot be
+        cross-checked, so the item reports the missing occurrence and nothing else.
+
+        The target here is real, so the only error that could appear alongside would
+        be an unwarranted "not the same occurrence" complaint.
+        """
+        real_target = Identification.objects.create(
+            occurrence=self.occurrences[1], taxon=self.taxon, user=self.superuser
+        )
+        response = self.post_bulk(
+            [{"occurrence_id": 9_999_997, "taxon_id": self.taxon.pk, "agreed_with_identification_id": real_target.pk}],
+            user=self.identifier,
+        )
+
+        errors = response.json()["results"][0]["errors"]
+        self.assertEqual(list(errors), ["occurrence_id"])
+
+    def test_an_item_reports_every_problem_it_has(self):
+        """Errors are reported per field, so one item with two problems names both."""
+        response = self.post_bulk(
+            [{"occurrence_id": 9_999_997, "taxon_id": 9_999_996}],
+            user=self.identifier,
+        )
+
+        errors = response.json()["results"][0]["errors"]
+        self.assertIn("occurrence_id", errors)
+        self.assertIn("taxon_id", errors)
+
+
+class TestBulkIdentificationValidation(BulkIdentificationTestCase):
+    def test_empty_batch_is_rejected(self):
+        response = self.post_bulk([], user=self.identifier)
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+
+    def test_missing_identifications_key_is_rejected(self):
+        self.client.force_authenticate(user=self.identifier)
+        response = self.client.post(ENDPOINT, {}, format="json")
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+
+    def test_non_integer_occurrence_id_returns_400_not_500(self):
+        self.client.force_authenticate(user=self.identifier)
+        response = self.client.post(
+            ENDPOINT, {"identifications": [{"occurrence_id": "abc", "taxon_id": self.taxon.pk}]}, format="json"
+        )
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+
+    def test_batch_larger_than_the_cap_is_rejected(self):
+        """
+        The cap is what fails an oversized batch, not some other rule.
+
+        Repeating one occurrence 201 times would be rejected by the duplicate check
+        instead, and unknown IDs would be reported per item, so both would pass this
+        test with the cap removed. Distinct IDs plus an assertion on the message pin
+        the cap itself.
+        """
+        items = [
+            {"occurrence_id": 9_000_000 + offset, "taxon_id": self.taxon.pk}
+            for offset in range(MAX_BULK_IDENTIFICATIONS + 1)
+        ]
+        response = self.post_bulk(items, user=self.identifier)
+
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertIn(str(MAX_BULK_IDENTIFICATIONS), str(response.json()))
+
+    def test_a_batch_at_the_cap_is_accepted(self):
+        """The cap rejects what is over it, not what is exactly at it."""
+        items = [
+            {"occurrence_id": 9_000_000 + offset, "taxon_id": self.taxon.pk}
+            for offset in range(MAX_BULK_IDENTIFICATIONS)
+        ]
+        response = self.post_bulk(items, user=self.identifier)
+
+        # Every occurrence is unknown, so each is reported as an error rather than
+        # the request being rejected outright.
+        self.assertEqual(response.status_code, status.HTTP_200_OK, response.content)
+        self.assertEqual(response.json()["error_count"], MAX_BULK_IDENTIFICATIONS)
+
+    def test_duplicate_occurrence_ids_are_rejected(self):
+        """
+        Two identifications for one occurrence in a single batch have no defined outcome.
+
+        Which one wins would depend on insert-order tiebreaks rather than on anything the
+        client asked for, so the batch is rejected instead.
+        """
+        response = self.post_bulk(
+            [self.item(self.occurrences[0]), self.item(self.occurrences[0])], user=self.identifier
+        )
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+
+    def test_batch_spanning_two_projects_is_rejected(self):
+        """A batch is authorized against a single project, so it must not span projects."""
+        other_project, other_deployment = setup_test_project(reuse=False)
+        create_taxa(project=other_project)
+        create_captures(deployment=other_deployment)
+        create_occurrences(deployment=other_deployment, num=1)
+        Identifier.assign_user(self.identifier, other_project)
+        other_occurrence = Occurrence.objects.filter(project=other_project).exclude(determination=None).first()
+        assert other_occurrence is not None
+
+        response = self.post_bulk([self.item(self.occurrences[0]), self.item(other_occurrence)], user=self.identifier)
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+
+    def test_withdrawn_cannot_be_set_through_the_bulk_endpoint(self):
+        """`withdrawn` is managed by the model; accepting it from a client would corrupt the invariant."""
+        response = self.post_bulk([self.item(self.occurrences[0], withdrawn=True)], user=self.identifier)
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK, response.content)
+        identification = Identification.objects.get(pk=response.json()["results"][0]["id"])
+        self.assertFalse(identification.withdrawn)
+
+
+class TestBulkIdentificationPermissions(BulkIdentificationTestCase):
+    """
+    The permission matrix.
+
+    A `detail=False` action is never routed through `has_object_permission`, so the
+    endpoint has to run the check itself. These cases fail loudly if it stops doing so.
+    """
+
+    def test_identifier_can_create(self):
+        response = self.post_bulk([self.item(self.occurrences[0])], user=self.identifier)
+        self.assertEqual(response.status_code, status.HTTP_200_OK, response.content)
+
+    def test_project_manager_can_create(self):
+        manager = User.objects.create_user(email="manager@insectai.org")  # type: ignore[attr-defined]
+        ProjectManager.assign_user(manager, self.project)
+        response = self.post_bulk([self.item(self.occurrences[0])], user=manager)
+        self.assertEqual(response.status_code, status.HTTP_200_OK, response.content)
+
+    def test_superuser_can_create(self):
+        response = self.post_bulk([self.item(self.occurrences[0])], user=self.superuser)
+        self.assertEqual(response.status_code, status.HTTP_200_OK, response.content)
+
+    def test_basic_member_is_forbidden(self):
+        """A project member without the identifier role must not be able to identify."""
+        response = self.post_bulk([self.item(self.occurrences[0])], user=self.basic_member)
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+        self.assertFalse(Identification.objects.filter(user=self.basic_member).exists())
+
+    def test_non_member_is_forbidden(self):
+        response = self.post_bulk([self.item(self.occurrences[0])], user=self.non_member)
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+        self.assertFalse(Identification.objects.filter(user=self.non_member).exists())
+
+    def test_anonymous_is_rejected(self):
+        response = self.client.post(ENDPOINT, {"identifications": [self.item(self.occurrences[0])]}, format="json")
+        self.assertIn(
+            response.status_code, (status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN), response.content
+        )
+        self.assertFalse(Identification.objects.exists())
+
+    def test_permission_is_denied_for_the_whole_batch(self):
+        """An unauthorized batch writes nothing at all, not a partial prefix."""
+        items = [self.item(occurrence) for occurrence in self.occurrences[:3]]
+        self.post_bulk(items, user=self.non_member)
+        self.assertFalse(Identification.objects.exists())
+
+
+class TestBulkIdentificationQueryCount(BulkIdentificationTestCase):
+    """
+    Query cost must stay linear in the batch, with a small and stable per-item slope.
+
+    Asserting one exact number for one batch size cannot distinguish fixed cost from
+    per-item cost, so it cannot catch an N+1 hidden in validation or in the response.
+    Measuring two batch sizes and comparing the slope can.
+    """
+
+    def measure(self, size: int) -> int:
+        # A fresh project per measurement keeps the two runs independent.
+        project, deployment = setup_test_project(reuse=False)
+        create_taxa(project=project)
+        create_captures(deployment=deployment)
+        create_occurrences(deployment=deployment, num=size)
+        Identifier.assign_user(self.identifier, project)
+        occurrences = list(Occurrence.objects.filter(project=project).exclude(determination=None))[:size]
+        taxon = Taxon.objects.exclude(pk=occurrences[0].determination_id).first()
+        assert taxon is not None
+        items = [{"occurrence_id": occurrence.pk, "taxon_id": taxon.pk} for occurrence in occurrences]
+
+        self.client.force_authenticate(user=self.identifier)
+        with CaptureQueriesContext(connection) as captured:
+            response = self.client.post(ENDPOINT, {"identifications": items}, format="json")
+        self.assertEqual(response.status_code, status.HTTP_200_OK, response.content)
+        self.assertEqual(response.json()["created_count"], size)
+        return len(captured.captured_queries)
+
+    def test_per_item_query_cost_stays_within_budget(self):
+        """
+        Each extra identification in a batch costs a bounded number of queries.
+
+        Measured at the time of writing: 8 queries per item, all of them inside
+        `Identification.save()` (withdraw previous, insert, then
+        `update_occurrence_determination` reading the current determination,
+        `best_identification` and `best_prediction` before saving the occurrence).
+        Resolving occurrences and taxa is batched and costs 2 queries for the whole
+        request, so it does not appear in this slope.
+
+        The budget is the measured cost, so any new per-item query fails this test.
+        If a change to the write path legitimately adds one, update the number here
+        deliberately rather than widening the budget to accommodate it.
+        """
+        small, large = 2, 6
+        queries_small = self.measure(small)
+        queries_large = self.measure(large)
+
+        slope = (queries_large - queries_small) / (large - small)
+        self.assertLessEqual(
+            slope,
+            8,
+            f"Each identification should cost a bounded number of queries, measured {slope:.1f} "
+            f"({queries_small} queries for {small} items, {queries_large} for {large}). "
+            f"A jump here usually means something started querying per item.",
+        )
diff --git a/docs/claude/planning/bulk-identifications-batched-write-followup.md b/docs/claude/planning/bulk-identifications-batched-write-followup.md
new file mode 100644
index 000000000..02e0bd3a2
--- /dev/null
+++ b/docs/claude/planning/bulk-identifications-batched-write-followup.md
@@ -0,0 +1,157 @@
+# Batched write path for bulk identifications — follow-up design
+
+**Status**: design only, not built. Follow-up to PR #1371.
+**Date**: 2026-07-17
+
+## Verdict
+
+**Do not build this yet.** The v1 loop (one `Identification.save()` per item inside a caught
+savepoint) is correct, reuses the only tested determination code, and costs ~8 queries per item on
+batches that are bounded by the UI page size (~25–100). A batched rewrite trades that safety for a
+saving that does not matter until batches grow past a page.
+
+**The trigger that should force it**: a "select all N occurrences matching this filter" mass-ID
+action, or raising `MAX_BULK_IDENTIFICATIONS` (`ami/main/api/serializers.py`) past ~500. At n≈100 the
+loop runs ~800 queries in one request; that is acceptable. At n=10 000 it is ~80 000 and holds row
+locks for the whole request — not acceptable. The number below is what changes the calculus.
+
+Rough query counts (read from code, **not measured**):
+
+| n | v1 loop | batched | note |
+|------|---------|---------|------|
+| 25 | ~200 | ~6 | loop fine |
+| 100 | ~800 | ~6 | loop fine |
+| 1 000 | ~8 000 | ~7 | loop starts to hurt |
+| 10 000 | ~80 000 | ~8 | loop unacceptable; batched needed |
+
+Batched cost is roughly constant because every step is a set operation; the loop is linear.
+
+## What the batched path must reproduce
+
+`Identification.save()` (`ami/main/models.py`) does three things per item:
+
+1. Withdraw the user's other non-withdrawn identifications on that occurrence
+   (`.filter(occurrence=, user=).exclude(pk=).update(withdrawn=True)`).
+2. Insert the row.
+3. `update_occurrence_determination(occurrence)` — recompute `determination` and
+   `determination_score`.
+
+The **score quirk is the spec**: `update_occurrence_determination` compares an int
+(`.values("determination")`) to a Taxon instance, so it always rewrites, and agreeing lifts the score
+to 1.0. A batched path must produce the identical result. Do not fix the quirk here (see the separate
+ticket below); a change to the score changes which occurrences pass score-threshold filters.
+
+## Algorithm sketch
+
+Order matters: **withdraw before insert**. You cannot exclude pks that do not exist yet, so withdraw
+the user's existing identifications on the affected occurrences first, then insert, then recompute.
+
+```python
+occ_ids = [item.occurrence_id for item in items]
+
+with transaction.atomic():
+    # 1. Withdraw the user's current identifications on these occurrences.
+    Identification.objects.filter(occurrence_id__in=occ_ids, user=user, withdrawn=False)\
+        .update(withdrawn=True)
+
+    # 2. Insert all new identifications at once.
+    created = Identification.objects.bulk_create([Identification(...) for item in items])
+
+    # 3. Recompute determination for every affected occurrence in a few set queries
+    #    (see below), then bulk_update the occurrences with update_fields.
+```
+
+Step 3, batched. The "best identification per occurrence" is a `DISTINCT ON`:
+
+```sql
+SELECT DISTINCT ON (occurrence_id) occurrence_id, id, taxon_id, 1.0 AS score
+FROM main_identification
+WHERE occurrence_id = ANY(%s) AND withdrawn = false
+ORDER BY occurrence_id, created_at DESC, id DESC   -- BEST_IDENTIFICATION_ORDER
+```
+
+Every affected occurrence now has at least one non-withdrawn identification (we just inserted one), so
+`best_prediction` never has to be consulted for this batch — that removes the one genuinely per-row
+query. `bulk_update(occurrences, ["determination_id", "determination_score"])` finishes it. Human
+score is always 1.0 (`Identification.score`), so the score column is a constant and needs no join.
+
+If a future batch can also *withdraw* identifications (this one does not), the "every occurrence has a
+non-withdrawn identification" assumption breaks and `best_prediction` returns — do not assume it away
+without checking.
+
+## Partial-failure reconciliation (the crux)
+
+`bulk_create` is all-or-nothing: it cannot report that row 37 of 200 failed. The v1 contract is
+per-row (save the good ones, report the bad). Options:
+
+1. **Validate everything up front, then one atomic batch.** Resolve occurrences/taxa/agree-targets and
+   reject any bad item *before* writing, so the only thing that can fail at write time is a concurrent
+   delete. Accept that a mid-flight concurrent delete turns into a whole-batch failure (rare). Cleanest
+   code; changes the contract from "always partial" to "partial only for pre-write validation errors".
+2. **Chunked batches with per-chunk savepoints.** `bulk_create` in chunks of, say, 100; a chunk that
+   raises falls back to the v1 per-item loop for that chunk only. Keeps the per-row contract exactly,
+   at some complexity. Best if the contract must not change.
+3. **Pre-flight existence check then batch, accepting the race.** Simplest, but the race window
+   (check → write) makes it the least honest.
+
+**Recommendation: option 2** if the per-row contract is load-bearing for the FE (it is today — the
+existing hook reports per-item results), option 1 if the FE is being rewritten anyway and a simpler
+"all valid or 400 with per-index errors" contract is acceptable. Decide *with the FE change*, not
+separately.
+
+## Divergence risk
+
+Introducing a batched recompute creates a second determination code path next to
+`update_occurrence_determination`. They will drift. The honest fix is to **express the scalar function
+in terms of the batch function**: `update_occurrence_determination(occurrence)` becomes
+`recompute_determinations([occurrence.pk])`, so there is one implementation and the single-item
+endpoint exercises the same code the batch does. This should land *before or with* the batched write,
+not after.
+
+## The test that must land first
+
+`update_occurrence_determination` has no tests today (`@TODO Add tests` in the source). Before swapping
+loop→batched, add a differential/property test that runs both paths over the same fixture matrix and
+asserts identical `(determination_id, determination_score, withdrawn flags)`:
+
+Fixture matrix (each row an occurrence state before the new identification):
+- agree with the current prediction
+- agree with an existing human identification
+- brand-new taxon (no prior human ID)
+- a prior identification by the **same** user (exercises withdraw-previous)
+- a prior identification by a **different** user (must survive)
+- an already-withdrawn identification present
+- no predictions at all
+- an occurrence with no detections
+- a duplicate occurrence in the batch (rejected upstream, but assert the recompute is not fed it)
+
+The property: for every state, loop-path result == batched-path result. That test is the gate; without
+it the swap is unverifiable.
+
+## Suggested follow-up tickets
+
+### Ticket A — batched write path
+
+> **Title**: Make bulk identifications scale to thousands of occurrences in one request
+>
+> Today an identifier can submit many identifications in one request, but each is written individually,
+> so a very large batch runs a query per occurrence and holds database locks for the whole request.
+> That is fine for the page-sized batches the interface sends now, but a future "identify everything
+> matching this filter" action would send thousands at once. This replaces the per-item write with a
+> set-based one (withdraw, insert, and recompute determinations in a handful of queries regardless of
+> batch size), gated behind a differential test that proves the batched recompute matches the existing
+> per-item behaviour exactly. Prerequisite: tests for `update_occurrence_determination`, which has none
+> today.
+
+### Ticket B — the score-comparison quirk
+
+> **Title**: Decide whether agreeing with a determination should raise its score to 1.0
+>
+> `update_occurrence_determination` compares a taxon ID to a taxon object, so its "nothing changed"
+> branch never runs and every identification rewrites the occurrence's determination score. In practice
+> this means agreeing with an existing determination lifts its score from the machine score to 1.0.
+> That may be the behaviour we want — a human-confirmed occurrence arguably should score 1.0 — but it
+> is currently reached by accident, and the score feeds the default score-threshold filters, so it
+> quietly affects which occurrences show up in filtered lists. This ticket is to decide the intended
+> behaviour and make the code express it on purpose, with a backfill plan if the decision changes
+> existing scores.
diff --git a/docs/claude/sessions/2026-07-17-bulk-identifications.md b/docs/claude/sessions/2026-07-17-bulk-identifications.md
new file mode 100644
index 000000000..9dbd0224e
--- /dev/null
+++ b/docs/claude/sessions/2026-07-17-bulk-identifications.md
@@ -0,0 +1,143 @@
+# Bulk identifications endpoint (PR #1371) — session notes
+
+**Date**: 2026-07-17
+**Branch**: `feat/bulk-identifications` (worktree off `origin/main`)
+**PR**: #1371 (draft) — *Let identifiers apply identifications to many occurrences in one request*
+**Author**: AI agent session
+
+Notes for whoever picks this up. Written for context recovery, not as a tutorial.
+
+## What shipped
+
+`POST /api/v2/identifications/bulk/` — a `detail=False` action on `IdentificationViewSet`
+(`ami/main/api/views.py`, search `def bulk`). Serializers are `Bulk*` in
+`ami/main/api/serializers.py`. Tests in `ami/main/test_bulk_identifications.py` (34).
+
+Contract: `{"identifications": [{occurrence_id, taxon_id, comment?, agreed_with_identification_id?,
+agreed_with_prediction_id?}]}` → `200 {created_count, error_count, results: [{index, occurrence_id,
+status, id?, errors?}]}`.
+
+## Findings that shaped the design (do not re-derive these)
+
+### The request shape was decided by the frontend, not by preference
+
+The mass-ID UI **already exists** and already fans out N POSTs
+(`ui/src/data-services/hooks/identifications/useCreateIdentifications.ts` — `Promise.allSettled`
+plus retry-only-the-rejected). Three call sites, all already building
+`IdentificationFieldValues[]`:
+
+- `ui/src/pages/occurrences/occurrences-actions.tsx` — "Agree/Confirm". **Heterogeneous**: each
+  item's taxon is *that occurrence's own* determination, and each item's agree target is its own
+  identification or prediction.
+- `ui/src/pages/occurrence-details/suggest-id/suggest-id.tsx` — one taxon, many occurrences.
+- `ui/src/pages/occurrence-details/id-quick-actions/id-button.tsx` — one taxon, many occurrences.
+
+The Agree flow makes `{taxon_id, occurrence_ids: []}` impossible. Hence a list of per-item objects.
+
+### Batch size is bounded by the page, not by the project
+
+`ui/src/pages/occurrences/occurrences.tsx` filters `selectedItems` to occurrences on the currently
+displayed page. So real batches are ~25–100. The ">100k occurrences" scaling worry does not reach
+this endpoint today. It would if a "select all matching this filter" action ever ships.
+
+### The task brief was wrong on two points
+
+- **PR #1281 "Update agree logic for human identifications" is a one-line frontend fix**
+  (`human-identification.tsx`, +1/−1). It is not backend agree logic.
+- **`user_agrees_with_identification` (`ami/main/models.py`) has zero call sites** — dead code. It
+  also carries an unbounded module-level `@functools.cache` keyed on model instances (stale results
+  + per-worker growth) if anyone ever wires it up.
+
+There is **no special backend agree handling**. `agreed_with_identification` / `agreed_with_prediction`
+are plain nullable FKs stored as provenance; `save()` never branches on them.
+
+### `update_occurrence_determination` compares an int to a Taxon (pre-existing)
+
+`current_determination` is read via `.values("determination")`, which yields the **raw FK id**, then
+compared against `top_identification.taxon`, a **model instance**. `Taxon != int` is always true, so
+the guard never short-circuits, the "no update needed" branch is effectively unreachable, and every
+identification rewrites `determination` and `determination_score`.
+
+**Verified empirically**, not inferred: a single POST agreeing with an existing determination moves
+`determination_score` from 0.9 to 1.0.
+
+Consequence: agreeing *does* set the score to 1.0. That may be desirable, but it is reached by
+accident. `determination_score` feeds project score-threshold filters
+(`ami/main/models_future/filters.py`), so changing it changes which occurrences appear in filtered
+lists. **Do not "fix" it as a drive-by.** The bulk endpoint deliberately reproduces today's behaviour.
+
+## Two bugs caught in review that are invisible in a diff
+
+### `ATOMIC_REQUESTS` defeats naive per-item transactions
+
+`config/settings/base.py:59` sets `ATOMIC_REQUESTS = True`. The whole request is therefore already
+one transaction, and a per-item `transaction.atomic()` is a **savepoint**, not an independent commit.
+Wrapping each item is not enough to get partial success: an exception escaping `save()` aborts the
+request transaction and discards every item already written, returning a 500.
+
+The fix is to **catch** the failure so the savepoint rolls back that item alone and the loop
+continues. Pinned by `test_a_database_failure_on_one_item_is_reported_without_losing_the_others`,
+which was confirmed to fail without the fix.
+
+### The action name silently becomes a permission nobody has
+
+A `detail=False` action is never routed through `has_object_permission`, and
+`ObjectPermission.has_permission` returns `True` unconditionally (`ami/base/permissions.py`), so the
+action must check permission itself. If it passes `view.action` (`"bulk"`) to `check_permission`, that
+misses the CRUD map in `BaseModel.check_permission` (`ami/base/models.py`) and falls through to
+`check_custom_permission` → `has_perm("bulk_identification", project)` — a permission no role grants.
+Superusers still pass via guardian's shortcut, so **a happy-path test written as a superuser would
+never catch it**. The action maps explicitly to `"create"`.
+
+Related: `Identification.check_permission`'s `"create"` branch depends only on the occurrence's
+project, which is why the batch is authorized once per project. A note now lives at
+`Identification.check_permission` (the line a future editor touches), not only at the endpoint.
+
+## Verification
+
+- 34 tests green (`ami.main.test_bulk_identifications`), full `ami.main` suite green.
+- **Live HTTP check against a real runserver** — the suite runs under `APITestCase`, which wraps each
+  test in a transaction, so `ATOMIC_REQUESTS` degrades to a savepoint there and real commit behaviour
+  is never exercised. The live check confirmed a mixed batch returns 200 with 2 created + 1 error, the
+  survivors **commit** (re-read from a separate process), the prior identification is withdrawn, and
+  403/400 paths behave. Scripts were kept out of the repo (throwaway).
+
+### Gotcha: the CI stack's *dev* database is polluted
+
+`docker compose -f docker-compose.ci.yml` → the `ami` database has a `main_project.license` NOT NULL
+column that **exists in no migration on `main`**, left behind by another branch. `migrate` reports
+"no migrations to apply" because the other branch's migration row is recorded. Seeding real data there
+fails with `null value in column "license"`.
+
+Workaround used: create a throwaway database, migrate it, run, drop it. Build the URL from
+`POSTGRES_*` env vars **inside** the container so no credential ever appears in a command.
+
+### Gotcha: two concurrent `--keepdb` runs corrupt each other
+
+Running two test commands against the CI stack at once produces spurious failures — they share the
+`test_ami` database. Also hit the known cachalot issue: `--keepdb` plus permission migrations →
+`IntegrityError on auth_group_permissions`. Fix is to drop the test DB **and** `redis-cli FLUSHALL`
+together (see `reference_ci_compose_cachalot_flush`).
+
+## Frontend switch (in this PR)
+
+The pressure relief only lands when the FE stops fanning out N POSTs. `useCreateIdentifications.ts` now
+makes one POST to `/bulk/`; the singular `useCreateIdentification.ts` (still used by the per-occurrence
+Agree card in `agree.tsx`) is untouched. The plural hook keeps its return shape
+(`isLoading`/`isSuccess`/`error`/`createIdentifications`), so the three callers (`occurrences-actions.tsx`,
+`suggest-id.tsx`, `id-button.tsx`) are unchanged. Two behaviours preserved on purpose: retry resends
+only the rejected items, and a partial failure is not reported as success (drives the Agree
+"Confirmed" state). "Select all" is page-bounded (`occurrence-gallery.tsx:173` maps the current page's
+`items`; default page size 20), so batches stay well under `MAX_BULK_IDENTIFICATIONS = 200`.
+
+FE verified: `tsc --noEmit`, eslint, prettier, and `yarn build` all clean. BE string-ID contract
+(`"123"` coerced by DRF IntegerField) pinned by `test_accepts_ids_sent_as_strings`.
+
+## Follow-ups
+
+- **Batched write path** — planned; the request contract does not change. See the planning doc.
+- **`update_occurrence_determination` int-vs-Taxon comparison** — needs a decision, then tests, before
+  anyone touches determination logic.
+- **`user_agrees_with_identification`** — delete or fix the `functools.cache`.
+- **Single-item endpoint asymmetries** — it accepts a writable `withdrawn` and unvalidated
+  `agreed_with_*` targets. The bulk endpoint rejects both.
diff --git a/ui/src/data-services/hooks/identifications/useCreateIdentification.ts b/ui/src/data-services/hooks/identifications/useCreateIdentification.ts
index a35fc1ad1..ace5c984c 100644
--- a/ui/src/data-services/hooks/identifications/useCreateIdentification.ts
+++ b/ui/src/data-services/hooks/identifications/useCreateIdentification.ts
@@ -5,7 +5,7 @@ import { getAuthHeader } from 'data-services/utils'
 import { useUser } from 'utils/user/userContext'
 import { IdentificationFieldValues } from './types'
 
-const convertToServerFieldValues = (
+export const convertToServerFieldValues = (
   fieldValues: IdentificationFieldValues
 ) => ({
   agreed_with_identification_id: fieldValues.agreeWith?.identificationId,
diff --git a/ui/src/data-services/hooks/identifications/useCreateIdentifications.ts b/ui/src/data-services/hooks/identifications/useCreateIdentifications.ts
index f7e64e04f..8a4073787 100644
--- a/ui/src/data-services/hooks/identifications/useCreateIdentifications.ts
+++ b/ui/src/data-services/hooks/identifications/useCreateIdentifications.ts
@@ -1,58 +1,104 @@
-import { useQueryClient } from '@tanstack/react-query'
-import { API_ROUTES, SUCCESS_TIMEOUT } from 'data-services/constants'
+import { useMutation, useQueryClient } from '@tanstack/react-query'
+import axios from 'axios'
+import { API_ROUTES, API_URL, SUCCESS_TIMEOUT } from 'data-services/constants'
+import { getAuthHeader } from 'data-services/utils'
 import { useEffect, useState } from 'react'
+import { useUser } from 'utils/user/userContext'
 import { IdentificationFieldValues } from './types'
-import { useCreateIdentification } from './useCreateIdentification'
+import { convertToServerFieldValues } from './useCreateIdentification'
 
+interface BulkIdentificationResult {
+  index: number
+  occurrence_id: number
+  status: 'created' | 'error'
+  id?: number
+  errors?: Record<string, string[]>
+}
+
+interface BulkIdentificationResponse {
+  created_count: number
+  error_count: number
+  results: BulkIdentificationResult[]
+}
+
+// Records the outcome of the last submission so a retry can resend only the
+// items that failed, and so the error message can report how many did.
+interface LastAttempt {
+  failed: IdentificationFieldValues[]
+  total: number
+}
+
+/**
+ * Apply an identification to many occurrences in a single request.
+ *
+ * The identifications are sent together to the bulk endpoint, which reports a
+ * result per item, rather than one request per occurrence. A partial failure
+ * (an occurrence deleted since the page loaded, say) leaves the successful items
+ * saved and lets a retry resend only the ones that failed.
+ */
 export const useCreateIdentifications = (
   occurrenceIds: string[],
   onSuccess?: () => void
 ) => {
+  const { user } = useUser()
   const queryClient = useQueryClient()
-  const [results, setResults] = useState<PromiseSettledResult<any>[]>()
-  const { createIdentification, isLoading, isSuccess, reset } =
-    useCreateIdentification(() => {
-      setTimeout(() => {
-        reset()
-      }, SUCCESS_TIMEOUT)
-    }, false)
-
-  const numRejected = results?.filter(
-    (result) => result.status === 'rejected'
-  ).length
-
-  const error = numRejected
-    ? results.length > 1
-      ? `${numRejected}/${results.length} updates were rejected, please retry.`
-      : 'The update was rejected, please retry.'
-    : undefined
+  const [lastAttempt, setLastAttempt] = useState<LastAttempt>()
 
+  const { mutateAsync, isLoading, isSuccess, isError, reset } = useMutation({
+    mutationFn: async (values: IdentificationFieldValues[]) => {
+      const { data } = await axios.post<BulkIdentificationResponse>(
+        `${API_URL}/${API_ROUTES.IDENTIFICATIONS}/bulk/`,
+        { identifications: values.map(convertToServerFieldValues) },
+        { headers: getAuthHeader(user) }
+      )
+      return { data, submitted: values }
+    },
+    onSuccess: ({ data, submitted }) => {
+      const failed = submitted.filter(
+        (_, index) => data.results[index]?.status === 'error'
+      )
+      setLastAttempt({ failed, total: submitted.length })
+      queryClient.invalidateQueries([API_ROUTES.IDENTIFICATIONS])
+      queryClient.invalidateQueries([API_ROUTES.OCCURRENCES])
+      onSuccess?.()
+      if (!failed.length) {
+        setTimeout(() => reset(), SUCCESS_TIMEOUT)
+      }
+    },
+  })
+
+  // Clear the retry state when the selection changes.
   useEffect(() => {
-    setResults(undefined)
+    setLastAttempt(undefined)
   }, [occurrenceIds.length])
 
+  const numRejected = lastAttempt?.failed.length
+  const partialError = numRejected
+    ? lastAttempt && lastAttempt.total > 1
+      ? `${numRejected}/${lastAttempt.total} updates were rejected, please retry.`
+      : 'The update was rejected, please retry.'
+    : undefined
+  // A rejected whole request (permission denied, invalid batch) surfaces here.
+  const requestError = isError
+    ? 'The update was rejected, please retry.'
+    : undefined
+  const error = partialError ?? requestError
+
   return {
+    // A partial failure is still a successful request, so only report success
+    // once every item landed.
+    isSuccess: isSuccess && !error,
     isLoading,
-    isSuccess,
     error,
     createIdentifications: async (params: IdentificationFieldValues[]) => {
-      const promises = params
-        .filter((_, index) => {
-          if (error) {
-            // Only retry rejected requests
-            return results?.[index]?.status === 'rejected'
-          }
-
-          return true
-        })
-        .map((variables) => createIdentification(variables))
-
-      setResults(undefined)
-      const result = await Promise.allSettled(promises)
-      setResults(result)
-      queryClient.invalidateQueries([API_ROUTES.IDENTIFICATIONS])
-      queryClient.invalidateQueries([API_ROUTES.OCCURRENCES])
-      onSuccess?.()
+      // On a retry, resend only the items that failed last time.
+      const toSubmit = partialError && lastAttempt ? lastAttempt.failed : params
+      try {
+        await mutateAsync(toSubmit)
+      } catch {
+        // A rejected request is surfaced through `error`; swallow it here so the
+        // caller's click handler does not see an unhandled rejection.
+      }
     },
   }
 }