diff --git a/src/backend/InvenTree/InvenTree/helpers_db.py b/src/backend/InvenTree/InvenTree/helpers_db.py
index 1d9b1ec328..5ed25e0bee 100644
--- a/src/backend/InvenTree/InvenTree/helpers_db.py
+++ b/src/backend/InvenTree/InvenTree/helpers_db.py
@@ -57,10 +57,16 @@ def bulk_create_and_fetch(
 
     instances = model.objects.filter(**lookup_filters)
 
-    pks = list(instances.values_list(id_field or 'pk', flat=True))
+    id_field = id_field or 'pk'
+
+    pks = list(instances.order_by(id_field).values_list(id_field, flat=True))
 
     # Override the metadata values to remove the temporary bulk_create_id
     instances.update(metadata=None)
 
     # Fetch the newly created items (by primary key, as the metadata filter no longer matches)
-    return model.objects.filter(**{f'{id_field or "pk"}__in': pks})
+    # Ordered by id_field: auto-increment values are assigned in insertion order on every
+    # backend, so this restores correspondence with the input 'items' list for callers that
+    # zip the result against it positionally. Without this, some backends (e.g. MySQL) may
+    # return 'pk__in' rows in a different order than they were inserted.
+    return model.objects.filter(**{f'{id_field}__in': pks}).order_by(id_field)
diff --git a/src/backend/InvenTree/InvenTree/serializers.py b/src/backend/InvenTree/InvenTree/serializers.py
index c96d726c2d..2c0662d86b 100644
--- a/src/backend/InvenTree/InvenTree/serializers.py
+++ b/src/backend/InvenTree/InvenTree/serializers.py
@@ -291,6 +291,110 @@ class FilterableSerializerMixin:
         return queryset
 
 
+@dataclass
+class PrefetchSpec:
+    """Describes a single bulk-prefetch to run against incoming (un-validated) request data.
+
+    Used by `BulkPrefetchSerializerMixin` to populate the {pk: instance} caches that
+    `InvenTree.fields.PrefetchedPrimaryKeyRelatedField` looks up against.
+
+    Attributes:
+        list_field: Name of the list field in the raw request data (e.g. 'items').
+        pk_field: Name of the pk-valued key within each entry of that list (e.g. 'stock_item').
+        queryset: Base queryset used to resolve the collected pks (e.g. `StockItem.objects.select_related(...)`).
+        cache_key: Context key to store the resulting {pk: instance} map under - must match
+            the `cache_key` passed to the corresponding `PrefetchedPrimaryKeyRelatedField`.
+    """
+
+    list_field: str
+    pk_field: str
+    queryset: QuerySet
+    cache_key: str
+
+
+class BulkPrefetchSerializerMixin:
+    """Mixin for a parent serializer whose nested list fields use PrefetchedPrimaryKeyRelatedField.
+
+    Without this mixin, PrefetchedPrimaryKeyRelatedField falls back to one .get() query per
+    list entry, since no cache has been populated in the context - for a request with
+    hundreds of entries, that turns validation itself into an O(n) query cost.
+
+    Subclasses declare `prefetch_fields`, a list of `PrefetchSpec` objects describing which
+    lists of raw entries to bulk-resolve before per-field validation runs:
+
+        class MySerializer(BulkPrefetchSerializerMixin, serializers.Serializer):
+            prefetch_fields = [
+                PrefetchSpec('items', 'stock_item', StockItem.objects.all(), '_stock_item'),
+                PrefetchSpec('items', 'build_line', BuildLine.objects.all(), '_build_line'),
+            ]
+
+            items = MyItemSerializer(many=True)
+
+        class MyItemSerializer(serializers.Serializer):
+            stock_item = PrefetchedPrimaryKeyRelatedField(
+                cache_key='_stock_item', queryset=StockItem.objects.all()
+            )
+            build_line = PrefetchedPrimaryKeyRelatedField(
+                cache_key='_build_line', queryset=BuildLine.objects.all()
+            )
+
+    Multiple specs may share the same `list_field` (as above), one per pk-valued key
+    that needs resolving within each entry.
+    """
+
+    prefetch_fields: list[PrefetchSpec] = []
+
+    @staticmethod
+    def _extract_pks(entries, key: str) -> set:
+        """Pull the raw pk values for 'key' out of a list of raw (un-validated) dicts.
+
+        Silently ignores entries which are not dicts, or whose value cannot be
+        interpreted as an integer pk - those are left for the per-field validation
+        (PrefetchedPrimaryKeyRelatedField) to reject with the usual error message.
+        """
+        pks = set()
+
+        for entry in entries or []:
+            if not isinstance(entry, dict):
+                continue
+
+            try:
+                pks.add(int(entry.get(key)))
+            except (TypeError, ValueError):
+                continue
+
+        return pks
+
+    def to_internal_value(self, data):
+        """Bulk-prefetch the objects referenced by each declared PrefetchSpec.
+
+        Populating context[cache_key] here means PrefetchedPrimaryKeyRelatedField resolves
+        each entry via an O(1) dict lookup, so the whole request is validated in a fixed,
+        small number of queries instead of one query per list entry.
+        """
+        if isinstance(data, dict):
+            for spec in self.prefetch_fields:
+                pks = self._extract_pks(data.get(spec.list_field), spec.pk_field)
+
+                self.context[spec.cache_key] = {
+                    obj.pk: obj for obj in spec.queryset.filter(pk__in=pks)
+                }
+
+            self.after_prefetch(data)
+
+        return super().to_internal_value(data)
+
+    def after_prefetch(self, data):
+        """Optional hook called once the pk caches are populated, before nested field validation runs.
+
+        Override this to bulk-compute anything else that per-entry `validate_<field>()`
+        or `validate()` methods would otherwise recompute one entry at a time - e.g. an
+        aggregate that would otherwise cost one query per entry. Stash the result in
+        `self.context` (the same mechanism `prefetch_fields` uses) so nested serializers
+        can look it up.
+        """
+
+
 class EmptySerializer(serializers.Serializer):
     """Empty serializer for use in testing."""
 
diff --git a/src/backend/InvenTree/build/models.py b/src/backend/InvenTree/build/models.py
index a15a78d8b8..74576ae06a 100644
--- a/src/backend/InvenTree/build/models.py
+++ b/src/backend/InvenTree/build/models.py
@@ -1,5 +1,6 @@
 """Build database model definitions."""
 
+import copy
 import decimal
 from typing import Optional, TypedDict
 
@@ -8,6 +9,7 @@ from django.core.exceptions import ValidationError
 from django.core.validators import MinValueValidator
 from django.db import models, transaction
 from django.db.models import F, Q, QuerySet, Sum
+from django.db.models.base import ModelState
 from django.db.models.functions import Coalesce
 from django.db.models.signals import post_save
 from django.dispatch.dispatcher import receiver
@@ -40,7 +42,9 @@ from common.models import ProjectCode
 from common.settings import get_global_setting
 from generic.enums import StringEnum
 from generic.states import StateTransitionMixin, StatusCodeMixin
-from plugin.events import trigger_event
+from InvenTree.helpers_db import bulk_create_and_fetch
+from plugin.events import bulk_trigger_event, trigger_event
+from stock.events import StockEvents
 from stock.status_codes import StockHistoryCode, StockStatus
 
 logger = structlog.get_logger('inventree')
@@ -648,13 +652,308 @@ class Build(
 
         return self.is_fully_allocated(tracked=False)
 
-    def complete_allocations(self, user) -> None:
+    @transaction.atomic
+    def complete_allocations(
+        self,
+        build_items: QuerySet,
+        quantities: Optional[dict] = None,
+        notes: Optional[str] = None,
+        user: Optional[User] = None,
+    ):
+        """Complete the allocation of stock for the given build lines.
+
+        Arguments:
+            build_items: QuerySet of build items to complete allocations for
+            quantities: Optional dict of quantities to allocate
+            notes: Optional notes for the allocation
+            user: The user completing the allocation
+
+        Notes:
+            This unrolls what would otherwise be a per-BuildItem call, so that the underlying StockItem,
+            StockItemTracking, BuildLine and BuildItem writes can be batched into
+            a handful of bulk queries instead of several per build item.
+        """
+        import part.tasks
+
+        notes = notes or ''
+        quantities = quantities or {}
+
+        # Lock this build's database row: all allocate / consume / complete
+        # operations for a given build are serialized against each other,
+        # so the in-memory BuildItem and BuildLine arithmetic below cannot
+        # race against a concurrent operation on the same build
+        Build.objects.select_for_update().get(pk=self.pk)
+
+        # Preselect related fields to avoid per-row database queries below
+        build_items = list(
+            build_items.select_related(
+                'build_line', 'stock_item', 'stock_item__part', 'install_into'
+            )
+        )
+
+        # Lock the distinct StockItem rows (in pk order, to avoid deadlocks) and
+        # re-read their quantities, so the clamping arithmetic below operates on
+        # the current committed values. Stock adjustments elsewhere lock these
+        # same rows, so concurrent adjustments cannot be silently overwritten.
+        locked_quantities = dict(
+            stock.models.StockItem.objects
+            .select_for_update()
+            .filter(pk__in={item.stock_item_id for item in build_items})
+            .order_by('pk')
+            .values_list('pk', 'quantity')
+        )
+
+        split_items = []  # (source_item, new_item, quantity) - stock to split off
+        install_items = []  # (build_item, target_item, output, quantity) - stock to install
+        consume_items = []  # (target_item, quantity) - stock to mark as consumed
+
+        # Canonical (mutable) copy of each distinct StockItem being drawn from - multiple
+        # build items may allocate against the same StockItem, so track running state
+        seen_stock_items: dict = {}
+
+        # Track the build items and build lines which have already been processed, to avoid double counting
+        seen_build_items: dict = {}
+        seen_build_lines: dict = {}
+
+        # Keep track of build items which need to be removed from the database after processing
+        # This is because the required quantity may be less than the allocated quantity, and we need to split the allocation
+        build_items_to_remove = set()
+
+        for build_item in build_items:
+            build_line = build_item.build_line
+
+            stock_item = (
+                seen_stock_items.get(build_item.stock_item_id) or build_item.stock_item
+            )
+
+            if (locked_qty := locked_quantities.pop(stock_item.pk, None)) is not None:
+                # First encounter: refresh the quantity from the locked database row
+                stock_item.quantity = locked_qty
+            elif stock_item.pk not in seen_stock_items:
+                # The stock item no longer exists in the database - skip
+                continue
+
+            quantity = quantities.get(build_item.pk, build_item.quantity)
+
+            if not isinstance(quantity, decimal.Decimal):
+                # Avoid float -> Decimal precision artifacts (e.g. Decimal(0.1))
+                quantity = decimal.Decimal(str(quantity))
+
+            # Clamp to the maximum quantity available for this build item and stock item
+            quantity = max(
+                decimal.Decimal(0),
+                min(quantity, build_item.quantity, stock_item.quantity),
+            )
+
+            if quantity <= 0:
+                continue
+
+            # Register this StockItem as "seen" now that we know it will be touched
+            seen_stock_items[stock_item.pk] = stock_item
+
+            trackable_install = stock_item.part.trackable and build_item.install_into_id
+            output = build_item.install_into if trackable_install else None
+
+            if quantity < stock_item.quantity:
+                # Split off exactly the consumed quantity into a new StockItem,
+                # leaving the remainder in place as available stock
+                new_item = copy.copy(stock_item)
+                new_item._state = ModelState()
+                new_item.pk = None
+                new_item.quantity = quantity
+                new_item.parent = stock_item
+
+                stock_item.quantity -= quantity
+
+                target_item = new_item
+                split_items.append((stock_item, new_item, quantity))
+            else:
+                target_item = stock_item
+
+            # Resolve the final resting state of the target item now - it may not have a
+            # primary key yet (if newly split), so this cannot depend on any pk lookups
+            target_item.consumed_by = self
+            target_item.location = None
+
+            if trackable_install:
+                target_item.belongs_to = output
+                install_items.append((build_item, target_item, output, quantity))
+            else:
+                consume_items.append((target_item, quantity))
+
+            # Increase the "consumed" quantity for the build line
+            _line = seen_build_lines.get(build_line.pk) or build_line
+            _line.consumed += quantity
+            seen_build_lines[build_line.pk] = _line
+
+            # Decrease the "quantity" for the build item
+            _item = seen_build_items.get(build_item.pk) or build_item
+            _item.quantity = max(decimal.Decimal(0), _item.quantity - quantity)
+            seen_build_items[build_item.pk] = _item
+
+            # If the total item quantity is now zero, we can remove the build item from the database
+            if _item.quantity <= 0:
+                build_items_to_remove.add(_item.pk)
+
+        # Nothing to do?
+        if not seen_build_items:
+            return
+
+        # Bulk-create the newly split-off stock items - this resolves their primary keys,
+        # which the tracking entries and (for installed items) BuildItem records need below
+        new_stock_item_data = [new_item for _, new_item, _ in split_items]
+
+        new_stock_items = bulk_create_and_fetch(
+            stock.models.StockItem, new_stock_item_data
+        )
+
+        # Back-populate the resolved primary keys onto the original 'new_item' objects
+        # (in-place, not via replacement) since 'install_items' / 'consume_items' hold
+        # references to these same objects (as 'target_item') and must see the
+        # resolved pk too. On backends that can't return pks from bulk_create (MySQL),
+        # 'new_stock_items' are freshly-fetched, distinct objects, so a plain list
+        # replacement here would leave those other references with pk=None.
+        for i, (_stock_item, new_item, _quantity) in enumerate(split_items):
+            new_item.pk = new_stock_items[i].pk
+
+        tracking_entries = []
+        split_events = []
+        install_events = []
+
+        # Split stock items for "split_items"
+        for source_item, new_item, quantity in split_items:
+            tracking_entries.append(
+                stock.models.StockItemTracking(
+                    item_id=new_item.pk,
+                    part_id=new_item.part_id,
+                    tracking_type=StockHistoryCode.SPLIT_FROM_PARENT.value,
+                    user=user,
+                    notes=notes,
+                    deltas={'stockitem': source_item.pk, 'quantity': float(quantity)},
+                )
+            )
+            tracking_entries.append(
+                stock.models.StockItemTracking(
+                    item_id=source_item.pk,
+                    part_id=source_item.part_id,
+                    tracking_type=StockHistoryCode.SPLIT_CHILD_ITEM.value,
+                    user=user,
+                    notes=notes,
+                    deltas={
+                        'removed': float(quantity),
+                        'quantity': float(source_item.quantity),
+                    },
+                )
+            )
+
+            split_events.append(((), {'id': new_item.pk, 'parent': source_item.pk}))
+
+        # Install stock items for "install_items"
+        for build_item, target_item, output, quantity in install_items:
+            tracking_entries.append(
+                stock.models.StockItemTracking(
+                    item_id=target_item.pk,
+                    part_id=target_item.part_id,
+                    tracking_type=StockHistoryCode.INSTALLED_INTO_ASSEMBLY.value,
+                    user=user,
+                    notes=notes,
+                    deltas={
+                        'stockitem': output.pk,
+                        'quantity': float(quantity),
+                        'buildorder': self.pk,
+                    },
+                )
+            )
+            tracking_entries.append(
+                stock.models.StockItemTracking(
+                    item_id=output.pk,
+                    part_id=output.part_id,
+                    tracking_type=StockHistoryCode.INSTALLED_CHILD_ITEM.value,
+                    user=user,
+                    notes=notes,
+                    deltas={'stockitem': target_item.pk, 'quantity': float(quantity)},
+                )
+            )
+
+            install_events.append((
+                (),
+                {'id': target_item.pk, 'assembly_id': output.pk},
+            ))
+
+            # Ensure the build item points to the (possibly newly split) stock item
+            build_item.stock_item = target_item
+
+        # Consume stock items for "consume_items"
+        for target_item, quantity in consume_items:
+            tracking_entries.append(
+                stock.models.StockItemTracking(
+                    item_id=target_item.pk,
+                    part_id=target_item.part_id,
+                    tracking_type=StockHistoryCode.BUILD_CONSUMED.value,
+                    user=user,
+                    notes=notes,
+                    deltas={'buildorder': self.pk, 'quantity': float(quantity)},
+                )
+            )
+
+        # Flush all StockItem field changes (quantity reductions, consumption, installs)
+        stock.models.StockItem.objects.bulk_update(
+            seen_stock_items.values(),
+            ['quantity', 'consumed_by', 'location', 'belongs_to'],
+        )
+
+        stock.models.StockItemTracking.objects.bulk_create(tracking_entries)
+
+        # Update build lines for "seen_build_lines"
+        BuildLine.objects.bulk_update(seen_build_lines.values(), ['consumed'])
+
+        # Update build items for "seen_build_items" (excluding those being removed)
+        build_items_to_update = [
+            item
+            for pk, item in seen_build_items.items()
+            if pk not in build_items_to_remove
+        ]
+        BuildItem.objects.bulk_update(build_items_to_update, ['quantity', 'stock_item'])
+
+        # Remove build items in "build_items_to_remove"
+        BuildItem.objects.filter(pk__in=build_items_to_remove).delete()
+
+        # Queue the ITEM_SPLIT / ITEM_INSTALLED_INTO_ASSEMBLY plugin events in bulk,
+        # rather than one offload_task() call (and one OrmQ insert) per item
+        bulk_trigger_event(StockEvents.ITEM_SPLIT, split_events)
+        bulk_trigger_event(StockEvents.ITEM_INSTALLED_INTO_ASSEMBLY, install_events)
+
+        # bulk_update()/bulk_create() above do not fire StockItem's post_save signal,
+        # which normally triggers a low-stock check for the affected part - so queue
+        # that check explicitly, once per distinct part touched by this call
+        touched_part_ids = {item.part_id for item in seen_stock_items.values()}
+
+        InvenTree.tasks.bulk_offload_task(
+            part.tasks.notify_low_stock_if_required,
+            [((part_id,), {}) for part_id in touched_part_ids],
+            group='notification',
+            force_async=True,
+        )
+
+    @transaction.atomic
+    def complete_outstanding_allocations(self, user) -> None:
         """Complete all stock allocations for this build order.
 
-        - This function is called when a build order is completed
+        - This function is called when a build order is completed.
+        - Any outstanding allocations will be consumed
+
+        Arguments:
+            user: The user who is completing the build
         """
-        # Remove untracked allocated stock
-        self.subtract_allocated_stock(user)
+        # Find all BuildItem objects which point to this build
+        items = self.allocated_stock.filter(
+            build_line__bom_item__sub_part__trackable=False
+        )
+
+        self.complete_allocations(build_items=items, user=user)
+
+        # Delete any remaining allocations
+        items.all().delete()
 
         # Ensure that there are no longer any BuildItem objects
         # which point to this Build Order
@@ -685,7 +984,7 @@ class Build(
         # from the database. Without this, two simultaneous completion requests
         # could both offload the completion task (duplicate events and
         # notifications; the stock-consuming side effects are themselves
-        # guarded against double-processing - see Build.subtract_allocated_stock()).
+        # guarded against double-processing
         self.status = Build.objects.select_for_update().get(pk=self.pk).status
 
         if self.status == BuildStatus.COMPLETE.value:
@@ -717,11 +1016,6 @@ class Build(
             group='build',
         )
 
-        self.completion_date = InvenTree.helpers.current_date()
-        self.completed_by = user
-        self.status = BuildStatus.COMPLETE.value
-        self.save()
-
     @transaction.atomic
     def issue_build(self):
         """Mark the Build as IN PRODUCTION.
@@ -810,7 +1104,7 @@ class Build(
         # from the database. Without this, two simultaneous cancellation requests
         # could both offload the cancellation task (duplicate events and
         # notifications; the stock-consuming side effects are themselves
-        # guarded against double-processing - see Build.subtract_allocated_stock()).
+        # guarded against double-processing
         self.status = Build.objects.select_for_update().get(pk=self.pk).status
 
         if self.status == BuildStatus.CANCELLED.value:
@@ -1055,29 +1349,6 @@ class Build(
         """Returns a QuerySet object of all BuildItem objects which point back to this Build."""
         return BuildItem.objects.filter(build_line__build=self)
 
-    @transaction.atomic
-    def subtract_allocated_stock(self, user) -> None:
-        """Removes the allocated untracked items from stock."""
-        # Find all BuildItem objects which point to this build
-        # Lock these rows (in pk order, to avoid deadlocks against other
-        # allocation operations), so a duplicated call to this method (e.g. a
-        # redelivered background task, or a completion and cancellation racing
-        # against each other) cannot process - and double-consume - the same
-        # allocations
-        items = list(
-            self.allocated_stock
-            .filter(build_line__bom_item__sub_part__trackable=False)
-            .select_for_update()
-            .order_by('pk')
-        )
-
-        # Remove stock
-        for item in items:
-            item.complete_allocation(user=user)
-
-        # Delete allocation
-        BuildItem.objects.filter(pk__in=[item.pk for item in items]).delete()
-
     @transaction.atomic
     def scrap_build_output(
         self, output: stock.models.StockItem, quantity, location, **kwargs
@@ -1133,12 +1404,11 @@ class Build(
         allocated_items = output.items_to_install.all()
 
         # Complete or discard allocations
-        for build_item in allocated_items:
-            if not discard_allocations:
-                build_item.complete_allocation(user=user)
+        if not discard_allocations:
+            self.complete_allocations(build_items=allocated_items, user=user)
 
         # Delete allocations
-        allocated_items.delete()
+        allocated_items.all().delete()
 
         output.add_tracking_entry(
             StockHistoryCode.BUILD_OUTPUT_REJECTED,
@@ -1255,9 +1525,7 @@ class Build(
             'stock_item', 'stock_item__part'
         )
 
-        for build_item in allocated_items:
-            # Complete the allocation of stock for that item
-            build_item.complete_allocation(user=user)
+        self.complete_allocations(build_items=allocated_items, user=user)
 
         # Delete the BuildItem objects from the database
         allocated_items.all().delete()
@@ -1294,6 +1562,75 @@ class Build(
         self.save(update_fields=['completed'])
         self.refresh_from_db(fields=['completed'])
 
+    @transaction.atomic
+    def allocate_stock(self, items: list[dict]):
+        """Allocated provided stock items against this Build.
+
+        Arguments:
+            items: A list of dictionaries containing the following keys:
+                - build_line: The BuildLine instance to allocate against
+                - stock_item: The StockItem instance to allocate
+                - quantity: The quantity to allocate
+                - output: Optional StockItem (build output) to install into
+
+        This function will create new BuildItem objects, or update existing ones
+        """
+        to_create = {}
+        to_update = {}
+
+        # Lock this build's database row: allocation requests are serialized
+        # against each other (and against consume / complete operations),
+        # so concurrent requests cannot both read the same starting quantity
+        Build.objects.select_for_update().get(pk=self.pk)
+
+        # Bulk-fetch any BuildItems which already exist for these (build_line, stock_item)
+        # pairs, keyed by (build_line_id, stock_item_id, install_into_id). Looking these up
+        # from a dict below avoids the one-query-per-item cost of a per-entry .filter().first()
+        # call, which otherwise dominates the query count for a large allocation request.
+        existing_items = {
+            (
+                existing.build_line_id,
+                existing.stock_item_id,
+                existing.install_into_id,
+            ): existing
+            for existing in BuildItem.objects.filter(
+                build_line__in={item['build_line'] for item in items},
+                stock_item__in={item['stock_item'] for item in items},
+            )
+        }
+
+        for item in items:
+            build_line = item['build_line']
+            stock_item = item['stock_item']
+            quantity = item['quantity']
+            output = item.get('output', None)
+
+            # Ignore allocation for consumable BOM items
+            if build_line.bom_item.is_consumable:
+                continue
+
+            filters = {
+                'build_line': build_line,
+                'stock_item': stock_item,
+                'install_into': output,
+            }
+
+            key = (build_line.pk, stock_item.pk, output.pk if output else None)
+
+            # If a BuildItem already exists for this allocation, update it
+            if build_item := existing_items.get(key):
+                build_item.quantity += quantity
+                to_update[build_item.pk] = build_item
+            elif build_item := to_create.get(key):
+                # Duplicate entry within this request - merge the quantities
+                build_item.quantity += quantity
+            else:
+                # This is a new BuildItem
+                to_create[key] = BuildItem(quantity=quantity, **filters)
+
+        BuildItem.objects.bulk_create(to_create.values(), batch_size=250)
+        BuildItem.objects.bulk_update(to_update.values(), ['quantity'], batch_size=250)
+
     @transaction.atomic
     def auto_allocate_stock(
         self, item_type: str = BuildItemTypes.UNTRACKED, **kwargs
@@ -1449,13 +1786,30 @@ class Build(
                 return 2
             return 3
 
-        new_items = []
-
         # Select only "untracked" line items
         untracked_lines = self.untracked_line_items.all()
+
         if line_ids:
             untracked_lines = untracked_lines.filter(pk__in=line_ids)
 
+        untracked_lines = untracked_lines.select_related(
+            'bom_item', 'bom_item__sub_part'
+        ).prefetch_related('bom_item__substitutes__part')
+
+        # Pre-compute the outstanding (unallocated) quantity for every line in a single
+        # query, instead of one aggregate query per line via BuildLine.unallocated_quantity()
+        untracked_lines = untracked_lines.annotate(
+            allocated=annotate_allocated_quantity(),
+            required=annotate_required_quantity(),
+        )
+
+        # First pass: resolve the candidate stock items for each line. We defer the actual
+        # allocation quantity math until every candidate stock item's outstanding allocated
+        # quantity can be resolved in a single bulk query (see below), rather than querying
+        # it (x3) per stock item as each line is processed.
+        line_data = []
+        candidate_stock = {}
+
         for line_item in untracked_lines:
             # Find the referenced BomItem
             bom_item = line_item.bom_item
@@ -1468,23 +1822,32 @@ class Build(
                 # User has specified that optional_items are to be ignored
                 continue
 
-            variant_parts = bom_item.sub_part.get_descendants(include_self=False)
-
-            unallocated_quantity = line_item.unallocated_quantity()
+            unallocated_quantity = max(line_item.required - line_item.allocated, 0)
 
             if unallocated_quantity <= 0:
                 # This BomItem is fully allocated, we can continue
                 continue
 
+            variant_parts = (
+                list(bom_item.sub_part.get_descendants(include_self=False))
+                if bom_item.allow_variants
+                else []
+            )
+
             # Check which parts we can "use" (may include variants and substitutes)
             available_parts = bom_item.get_valid_parts_for_allocation(
-                allow_variants=True, allow_inactive=False, allow_substitutes=substitutes
+                allow_variants=True,
+                allow_inactive=False,
+                allow_substitutes=substitutes,
+                variant_parts=variant_parts,
             )
 
             # Look for available stock items
-            available_stock = stock.models.StockItem.objects.filter(
-                stock.models.StockItem.IN_STOCK_FILTER
-            )
+            # select_related('part') avoids a per-item query in the stock_sort() key
+            # function below, which inspects item.part for every candidate.
+            available_stock = stock.models.StockItem.objects.select_related(
+                'part'
+            ).filter(stock.models.StockItem.IN_STOCK_FILTER)
 
             # Filter by list of available parts
             available_stock = available_stock.filter(part__in=list(available_parts))
@@ -1535,42 +1898,56 @@ class Build(
                 key=lambda item, b=bom_item, v=variant_parts: stock_sort(item, b, v),
             )
 
-            if len(available_stock) == 1 or interchangeable:
-                # Either there is only a single stock item available,
-                # or all items are "interchangeable" and we don't care where we take stock from
+            if len(available_stock) != 1 and not interchangeable:
+                # Multiple stock items are available, but they are not interchangeable -
+                # the user must manually decide how to allocate them.
+                continue
 
-                for stock_item in available_stock:
-                    # Skip inactive parts
-                    if not stock_item.part.active:
-                        continue
+            for stock_item in available_stock:
+                candidate_stock[stock_item.pk] = stock_item
 
-                    # How much of the stock item is "available" for allocation?
-                    quantity = min(
-                        unallocated_quantity, stock_item.unallocated_quantity()
-                    )
+            line_data.append((line_item, unallocated_quantity, available_stock))
 
-                    if quantity > 0:
-                        try:
-                            new_items.append(
-                                BuildItem(
-                                    build_line=line_item,
-                                    stock_item=stock_item,
-                                    quantity=quantity,
-                                )
+        # Bulk-resolve the outstanding allocated quantity for every candidate stock item
+        # identified above, in a fixed number of queries rather than 3 queries per item.
+        allocated_by_pk = stock.models.StockItem.bulk_allocation_count(
+            candidate_stock.values()
+        )
+
+        new_items = []
+
+        for line_item, unallocated_quantity, available_stock in line_data:
+            # Either there is only a single stock item available, or all items are
+            # "interchangeable" and we don't care where we take stock from
+            for stock_item in available_stock:
+                # How much of the stock item is "available" for allocation?
+                stock_unallocated = max(
+                    stock_item.quantity - allocated_by_pk.get(stock_item.pk, 0), 0
+                )
+                quantity = min(unallocated_quantity, stock_unallocated)
+
+                if quantity > 0:
+                    try:
+                        new_items.append(
+                            BuildItem(
+                                build_line=line_item,
+                                stock_item=stock_item,
+                                quantity=quantity,
                             )
+                        )
 
-                            # Subtract the required quantity
-                            unallocated_quantity -= quantity
+                        # Subtract the required quantity
+                        unallocated_quantity -= quantity
 
-                        except (ValidationError, serializers.ValidationError) as exc:
-                            # Re-raise with a Django-compatible validation payload
-                            raise ValidationError(
-                                serializers.as_serializer_error(exc)
-                            ) from exc
+                    except (ValidationError, serializers.ValidationError) as exc:
+                        # Re-raise with a Django-compatible validation payload
+                        raise ValidationError(
+                            serializers.as_serializer_error(exc)
+                        ) from exc
 
-                    if unallocated_quantity <= 0:
-                        # We have now fully-allocated this BomItem - no need to continue!
-                        break
+                if unallocated_quantity <= 0:
+                    # We have now fully-allocated this BomItem - no need to continue!
+                    break
 
         # Bulk-create the new BuildItem objects
         BuildItem.objects.bulk_create(new_items, batch_size=250)
@@ -2072,89 +2449,26 @@ class BuildItem(InvenTree.models.InvenTreeMetadataModel):
         """Return the BomItem associated with this BuildItem."""
         return self.build_line.bom_item if self.build_line else None
 
-    @transaction.atomic
-    def complete_allocation(self, quantity=None, notes: str = '', user=None) -> None:
+    def complete_allocation(
+        self, quantity: Optional[decimal.Decimal] = None, notes: str = '', user=None
+    ) -> None:
         """Complete the allocation of this BuildItem into the output stock item.
 
+        A thin wrapper around Build.complete_allocations(), for completing a single BuildItem.
+
         Arguments:
             quantity: The quantity to allocate (default is the full quantity)
             notes: Additional notes to add to the transaction
             user: The user completing the allocation
-
-        - If the referenced part is trackable, the stock item will be *installed* into the build output
-        - If the referenced part is *not* trackable, the stock item will be *consumed* by the build order
-
-        TODO: This is quite expensive (in terms of number of database hits) - and requires some thought
-        TODO: Revisit, and refactor!
-
         """
-        # If the quantity is not provided, use the quantity of this BuildItem
-        if quantity is None:
-            quantity = self.quantity
+        quantities = {self.pk: quantity} if quantity is not None else None
 
-        item = self.stock_item
-
-        # Lock the stock item's row and refresh its quantity, so the clamp and
-        # split decision below operate on the current committed quantity rather
-        # than a stale in-memory copy (concurrent allocations may have consumed
-        # stock from this same item in the meantime)
-        if not item.lock_quantity():
-            # The stock item no longer exists - remove this (now invalid) allocation
-            self.delete()
-            return
-
-        # Ensure we are not allocating more than available
-        if quantity > item.quantity:
-            quantity = item.quantity
-
-        if quantity <= 0:
-            # There is nothing to consume or install:
-            # simply remove this (empty) allocation
-            self.delete()
-            return
-
-        # Split the allocated stock if there are more available than allocated
-        if item.quantity > quantity:
-            item = item.splitStock(quantity, None, user, notes=notes)
-
-        # For a trackable part, special consideration needed!
-        if item.part.trackable:
-            # Make sure we are pointing to the new item
-            self.stock_item = item
-            self.save()
-
-            # Install the stock item into the output
-            self.install_into.installStockItem(
-                item, quantity, user, notes, build=self.build
-            )
-
-        else:
-            # Mark the item as "consumed" by the build order
-            item.consumed_by = self.build
-            item.location = None
-            item.save(add_note=False)
-
-            item.add_tracking_entry(
-                StockHistoryCode.BUILD_CONSUMED,
-                user,
-                notes=notes,
-                deltas={'buildorder': self.build.pk, 'quantity': float(item.quantity)},
-            )
-
-        # Increase the "consumed" count for the associated BuildLine
-        # Increment at the database level to prevent lost updates
-        # (multiple allocations against the same BuildLine may complete concurrently)
-        self.build_line.consumed = F('consumed') + quantity
-        self.build_line.save(update_fields=['consumed'])
-        self.build_line.refresh_from_db(fields=['consumed'])
-
-        # Decrease the allocated quantity
-        self.quantity = max(0, self.quantity - quantity)
-
-        if self.quantity <= 0:
-            self.delete()
-        else:
-            self.save()
+        self.build.complete_allocations(
+            BuildItem.objects.filter(pk=self.pk),
+            quantities=quantities,
+            notes=notes,
+            user=user,
+        )
 
     build_line = models.ForeignKey(
         BuildLine, on_delete=models.CASCADE, null=True, related_name='allocations'
diff --git a/src/backend/InvenTree/build/serializers.py b/src/backend/InvenTree/build/serializers.py
index 004a47bef5..951d5de48e 100644
--- a/src/backend/InvenTree/build/serializers.py
+++ b/src/backend/InvenTree/build/serializers.py
@@ -31,8 +31,10 @@ import part.serializers as part_serializers
 import stock.models as stock_models
 from common.settings import get_global_setting
 from generic.states.fields import InvenTreeCustomStatusSerializerMixin
+from InvenTree.fields import PrefetchedPrimaryKeyRelatedField
 from InvenTree.mixins import DataImportExportSerializerMixin
 from InvenTree.serializers import (
+    BulkPrefetchSerializerMixin,
     CustomStatusSerializerMixin,
     DuplicateOptionsSerializer,
     FilterableSerializerMixin,
@@ -41,6 +43,7 @@ from InvenTree.serializers import (
     InvenTreeTaggitSerializer,
     NotesFieldMixin,
     OptionalField,
+    PrefetchSpec,
 )
 from stock.generators import generate_batch_code
 from stock.models import StockItem, StockLocation
@@ -858,7 +861,8 @@ class BuildAllocationItemSerializer(serializers.Serializer):
 
         fields = ['build_item', 'stock_item', 'quantity', 'output']
 
-    build_line = serializers.PrimaryKeyRelatedField(
+    build_line = PrefetchedPrimaryKeyRelatedField(
+        cache_key='_build_line',
         queryset=BuildLine.objects.all(),
         many=False,
         allow_null=False,
@@ -886,7 +890,8 @@ class BuildAllocationItemSerializer(serializers.Serializer):
 
         return build_line
 
-    stock_item = serializers.PrimaryKeyRelatedField(
+    stock_item = PrefetchedPrimaryKeyRelatedField(
+        cache_key='_stock_item',
         queryset=StockItem.objects.all(),
         many=False,
         allow_null=False,
@@ -912,7 +917,8 @@ class BuildAllocationItemSerializer(serializers.Serializer):
 
         return quantity
 
-    output = serializers.PrimaryKeyRelatedField(
+    output = PrefetchedPrimaryKeyRelatedField(
+        cache_key='_output',
         queryset=StockItem.objects.filter(is_building=True),
         many=False,
         allow_null=True,
@@ -931,11 +937,26 @@ class BuildAllocationItemSerializer(serializers.Serializer):
 
         # build = self.context['build']
 
-        # TODO: Check that the "stock item" is valid for the referenced "sub_part"
-        # Note: Because of allow_variants options, it may not be a direct match!
+        # Check that the "stock item" is valid for the referenced "sub_part"
+        # Fast path: direct part match, avoids querying substitutes / variants below
+        # (which would otherwise cost extra queries per line for bulk allocations)
+        if stock_item.part_id != build_line.bom_item.sub_part_id and (
+            not build_line.bom_item.is_stock_item_valid(stock_item)
+        ):
+            raise ValidationError({
+                'stock_item': _('Selected stock item does not match BOM line')
+            })
 
         # Check that the quantity does not exceed the available amount from the stock item
-        q = stock_item.unallocated_quantity()
+        allocated = self.context.get('_stock_item_allocated')
+
+        if allocated is not None:
+            q = max(
+                stock_item.quantity - allocated.get(stock_item.pk, Decimal(0)),
+                Decimal(0),
+            )
+        else:
+            q = stock_item.unallocated_quantity()
 
         if quantity > q:
             q = InvenTree.helpers.clean_decimal(q)
@@ -961,7 +982,7 @@ class BuildAllocationItemSerializer(serializers.Serializer):
         return data
 
 
-class BuildAllocationSerializer(serializers.Serializer):
+class BuildAllocationSerializer(BulkPrefetchSerializerMixin, serializers.Serializer):
     """Serializer for allocating stock items against a build order."""
 
     class Meta:
@@ -969,8 +990,33 @@ class BuildAllocationSerializer(serializers.Serializer):
 
         fields = ['items']
 
+    prefetch_fields = [
+        PrefetchSpec(
+            'items',
+            'build_line',
+            BuildLine.objects.select_related(
+                'build', 'bom_item__part', 'bom_item__sub_part'
+            ),
+            '_build_line',
+        ),
+        PrefetchSpec('items', 'stock_item', StockItem.objects.all(), '_stock_item'),
+        PrefetchSpec(
+            'items', 'output', StockItem.objects.filter(is_building=True), '_output'
+        ),
+    ]
+
     items = BuildAllocationItemSerializer(many=True)
 
+    def after_prefetch(self, data):
+        """Bulk-compute allocated quantities for every referenced stock item.
+
+        BuildAllocationItemSerializer.validate() checks each stock item's unallocated.
+        """
+        stock_items = self.context.get('_stock_item', {}).values()
+        self.context['_stock_item_allocated'] = StockItem.bulk_allocation_count(
+            stock_items
+        )
+
     def validate(self, data):
         """Validation."""
         data = super().validate(data)
@@ -987,42 +1033,8 @@ class BuildAllocationSerializer(serializers.Serializer):
         data = self.validated_data
 
         items = data.get('items', [])
-
-        with transaction.atomic():
-            for item in items:
-                build_line = item['build_line']
-                stock_item = item['stock_item']
-                quantity = item['quantity']
-                output = item.get('output', None)
-
-                # Ignore allocation for consumable BOM items
-                if build_line.bom_item.is_consumable:
-                    continue
-
-                params = {
-                    'build_line': build_line,
-                    'stock_item': stock_item,
-                    'install_into': output,
-                }
-
-                try:
-                    # Lock the row, so concurrent allocations cannot both read
-                    # the same starting quantity (lost update)
-                    if build_item := (
-                        BuildItem.objects.select_for_update().filter(**params).first()
-                    ):
-                        # Find an existing BuildItem for this stock item
-                        # If it exists, increase the quantity
-                        build_item.quantity += quantity
-                        build_item.save()
-                    else:
-                        # Create a new BuildItem to allocate stock
-                        build_item = BuildItem.objects.create(
-                            quantity=quantity, **params
-                        )
-                except (ValidationError, DjangoValidationError) as exc:
-                    # Catch model errors and re-throw as DRF errors
-                    raise ValidationError(detail=serializers.as_serializer_error(exc))
+        build = self.context['build']
+        build.allocate_stock(items)
 
 
 class BuildAutoAllocationSerializer(serializers.Serializer):
diff --git a/src/backend/InvenTree/build/tasks.py b/src/backend/InvenTree/build/tasks.py
index 13355ac043..ad90443493 100644
--- a/src/backend/InvenTree/build/tasks.py
+++ b/src/backend/InvenTree/build/tasks.py
@@ -6,6 +6,7 @@ from typing import Optional
 
 from django.contrib.auth.models import User
 from django.db import transaction
+from django.db.models import Q
 from django.utils.translation import gettext_lazy as _
 
 import structlog
@@ -49,7 +50,7 @@ def consume_build_stock(
         items: Optional dict of BuildItem IDs (and quantities)to consume
         user_id: The ID of the user who initiated the stock consumption
     """
-    from build.models import Build, BuildItem, BuildLine
+    from build.models import Build, BuildItem
 
     build = Build.objects.get(pk=build_id)
     user = User.objects.filter(pk=user_id).first() if user_id else None
@@ -58,24 +59,19 @@ def consume_build_stock(
     items = items or {}
     notes = kwargs.pop('notes', '')
 
-    # Extract the relevant BuildLine and BuildItem objects
-    with transaction.atomic():
-        # Consume each of the specified BuildLine objects
-        for line_id in lines:
-            if build_line := BuildLine.objects.filter(pk=line_id, build=build).first():
-                for item in build_line.allocations.all():
-                    item.complete_allocation(
-                        quantity=item.quantity, notes=notes, user=user
-                    )
+    # Condense the provided lines and items into a single BuildItem queryset,
+    # preselecting the related StockItem to avoid per-item queries downstream
+    build_items = (
+        BuildItem.objects
+        .filter(
+            Q(build_line__pk__in=lines) | Q(pk__in=items.keys()),
+            build_line__build=build,
+        )
+        .select_related('stock_item', 'stock_item__part')
+        .distinct()
+    )
 
-        # Consume each of the specified BuildItem objects
-        for item_id, quantity in items.items():
-            if build_item := BuildItem.objects.filter(
-                pk=item_id, build_line__build=build
-            ).first():
-                build_item.complete_allocation(
-                    quantity=quantity, notes=notes, user=user
-                )
+    build.complete_allocations(build_items, quantities=items, notes=notes, user=user)
 
 
 @tracer.start_as_current_span('complete_build_allocations')
@@ -97,7 +93,7 @@ def complete_build_allocations(build_id: int, user_id: int):
     else:
         user = None
 
-    build_order.complete_allocations(user)
+    build_order.complete_outstanding_allocations(user)
 
 
 @tracer.start_as_current_span('delete_build_outputs')
@@ -303,15 +299,34 @@ def complete_build(build_id: int, user_id: int, trim_allocated_stock: bool = Fal
         trim_allocated_stock: If True, trim any allocated stock which was not consumed
     """
     from build.models import Build
+    from build.status_codes import BuildStatus
 
-    build = Build.objects.get(pk=build_id)
-    user = User.objects.filter(pk=user_id).first() if user_id else None
+    with transaction.atomic():
+        # Lock the build row: concurrent completion tasks (duplicate task
+        # delivery, or repeated completion requests while the build is still
+        # IN PRODUCTION) are serialized, and the status is re-checked below
+        build = Build.objects.select_for_update().get(pk=build_id)
 
-    if trim_allocated_stock:
-        build.trim_allocated_stock()
+        if build.status == BuildStatus.COMPLETE.value:
+            logger.warning(
+                'Build order <%s> is already complete - skipping completion task',
+                build.pk,
+            )
+            return
 
-    # Complete any remaining allocations for this build order
-    complete_build_allocations(build_id, user_id)
+        user = User.objects.filter(pk=user_id).first() if user_id else None
+
+        if trim_allocated_stock:
+            build.trim_allocated_stock()
+
+        # Complete any remaining allocations for this build order
+        complete_build_allocations(build_id, user_id)
+
+        # Mark the build as completed
+        build.completion_date = InvenTree.helpers.current_date()
+        build.completed_by = user
+        build.status = BuildStatus.COMPLETE.value
+        build.save()
 
     # Register an event
     trigger_event(BuildEvents.COMPLETED, id=build.pk)
diff --git a/src/backend/InvenTree/build/test_api.py b/src/backend/InvenTree/build/test_api.py
index 2b9c6d9e3e..f4b993712e 100644
--- a/src/backend/InvenTree/build/test_api.py
+++ b/src/backend/InvenTree/build/test_api.py
@@ -3,8 +3,10 @@
 from datetime import datetime, timedelta
 from typing import Optional
 
+from django.db.models import Sum
 from django.urls import reverse
 
+from django_q.models import OrmQ
 from rest_framework import status
 
 from build.models import Build, BuildItem, BuildLine
@@ -12,8 +14,9 @@ from build.status_codes import BuildStatus
 from common.settings import set_global_setting
 from InvenTree.unit_test import InvenTreeAPITestCase
 from part.models import BomItem, BomItemSubstitute, Part, PartTestTemplate
-from stock.models import StockItem, StockLocation, StockSortOrder
-from stock.status_codes import StockStatus
+from stock.events import StockEvents
+from stock.models import StockItem, StockItemTracking, StockLocation, StockSortOrder
+from stock.status_codes import StockHistoryCode, StockStatus
 
 
 class TestBuildAPI(InvenTreeAPITestCase):
@@ -687,7 +690,7 @@ class BuildAllocationTest(BuildAPITest):
         wrong_line = None
 
         for line in lines:
-            if line.bom_item.sub_part.pk != si.pk:
+            if line.bom_item.sub_part.pk != si.part.pk:
                 wrong_line = line
                 break
 
@@ -2147,6 +2150,432 @@ class BuildConsumeTest(BuildAPITest):
         for line in self.build.build_lines.all():
             self.assertEqual(line.consumed, 100)
 
+    def test_consume_many_lines(self):
+        """Test consuming stock against a build order with a large number of BOM lines.
+
+        - Create 100 sub-components for a new assembly
+        - Create a build order against the assembly
+        - Create a stock item for each component - some with exactly the required
+          quantity, others with more than required (to test stock splitting)
+        - Every 10th line is instead allocated from *two* different stock items,
+          whose quantities combined equal the required quantity (to test multiple
+          BuildItems being allocated against a single BuildLine)
+        - Allocate stock against each build line
+        - Complete the build order (via output creation, output completion, and finish)
+        - Check that every build line is fully consumed
+        - Check that stock items are reduced (and split) correctly
+        - Check that associated events are triggered
+        - Check that required low-stock checks are triggered for each component
+        - Check that the correct stock tracking entries are created for each split stock item
+        """
+        N = 100
+        bom_quantity = 5
+        build_quantity = 10
+        required_quantity = bom_quantity * build_quantity
+
+        # Every MULTI_ITEM_STRIDE'th line is allocated from two stock items instead of one
+        MULTI_ITEM_STRIDE = 10
+
+        assembly = Part.objects.create(
+            name='Large Test Assembly',
+            description='Assembly with a large number of BOM lines',
+            assembly=True,
+        )
+
+        components = [
+            Part.objects.create(
+                name=f'Large Test Component {i}',
+                description=f'Large Test Component Description {i}',
+                component=True,
+            )
+            for i in range(N)
+        ]
+
+        for component in components:
+            BomItem.objects.create(
+                part=assembly, sub_part=component, quantity=bom_quantity
+            )
+
+        build = Build.objects.create(
+            part=assembly,
+            reference='BO-12350',
+            quantity=build_quantity,
+            title='Large Test Build',
+        )
+
+        self.assertEqual(build.build_lines.count(), N)
+
+        # Create stock item(s) for each component:
+        # - Every MULTI_ITEM_STRIDE'th line gets *two* stock items, whose quantities
+        #   combined equal the required quantity (no splitting needed for either)
+        # - Of the rest, odd-indexed components get more stock than is required (to
+        #   test splitting), even-indexed components get exactly the required quantity
+        stock_items = []
+
+        for idx, component in enumerate(components):
+            if idx % MULTI_ITEM_STRIDE == 0:
+                first_quantity = required_quantity * 3 // 5
+                stock_items.append([
+                    StockItem.objects.create(part=component, quantity=first_quantity),
+                    StockItem.objects.create(
+                        part=component, quantity=required_quantity - first_quantity
+                    ),
+                ])
+            else:
+                stock_items.append([
+                    StockItem.objects.create(
+                        part=component,
+                        quantity=required_quantity + (25 if idx % 2 else 0),
+                    )
+                ])
+
+        # Starting point (before splitting stock)
+        N_STOCK_ITEMS = StockItem.objects.count()
+
+        # Allocate stock against each build line - lines with multiple stock items get
+        # one allocation entry per stock item, with quantities summing to the required amount
+        allocation_items = []
+
+        for line, items in zip(build.build_lines.all(), stock_items, strict=True):
+            if len(items) > 1:
+                for si in items:
+                    allocation_items.append({
+                        'build_line': line.pk,
+                        'stock_item': si.pk,
+                        'quantity': si.quantity,
+                    })
+            else:
+                allocation_items.append({
+                    'build_line': line.pk,
+                    'stock_item': items[0].pk,
+                    'quantity': required_quantity,
+                })
+
+        data = {'items': allocation_items}
+
+        self.post(
+            reverse('api-build-allocate', kwargs={'pk': build.pk}),
+            data,
+            expected_code=201,
+            benchmark=True,
+            max_query_time=1.0,
+            max_query_count=50,
+        )
+
+        n_multi_item_lines = len(range(0, N, MULTI_ITEM_STRIDE))
+        self.assertEqual(build.allocated_stock.count(), N + n_multi_item_lines)
+        self.assertTrue(build.are_untracked_parts_allocated)
+
+        # Create (and complete) a single build output for the full build quantity
+        build.create_build_output(build_quantity)
+        output = build.incomplete_outputs.first()
+
+        self.post(
+            reverse('api-build-output-complete', kwargs={'pk': build.pk}),
+            {
+                'outputs': [{'output': output.pk}],
+                'location': 1,
+                'status': StockStatus.OK.value,
+            },
+            expected_code=200,
+            benchmark=True,
+            max_query_time=0.5,
+            max_query_count=200,
+        )
+
+        build.refresh_from_db()
+        self.assertEqual(build.completed, build_quantity)
+        self.assertTrue(build.can_complete)
+
+        # Enable plugin events, and queue them (rather than firing synchronously),
+        # so we can inspect exactly what was queued once the build is finished
+        set_global_setting('ENABLE_PLUGINS_EVENTS', True, change_user=None)
+
+        # Start with a fresh slate for the OrmQ queue, so we can inspect exactly what is queued during the build finish
+        OrmQ.objects.all().delete()
+
+        # Finish the build order - this consumes all allocated stock
+        with self.settings(
+            PLUGIN_TESTING_EVENTS=True, PLUGIN_TESTING_EVENTS_ASYNC=True
+        ):
+            self.post(
+                reverse('api-build-finish', kwargs={'pk': build.pk}),
+                {},
+                expected_code=201,
+                benchmark=True,
+                max_query_count=250,
+                max_query_time=1.0,
+            )
+
+        build.refresh_from_db()
+        self.assertTrue(build.is_complete)
+
+        queued_tasks = list(OrmQ.objects.all())
+
+        # register_event tasks queued for StockEvents.ITEM_SPLIT, one per split component
+        split_event_tasks = [
+            task
+            for task in queued_tasks
+            if task.func() == 'plugin.base.event.events.register_event'
+            and task.args() == (StockEvents.ITEM_SPLIT,)
+        ]
+        self.assertEqual(len(split_event_tasks), N // 2)
+
+        # 'notify_low_stock_if_required' should be queued once per distinct part touched -
+        # every one of the N components here is a distinct part
+        low_stock_tasks = [
+            task
+            for task in queued_tasks
+            if task.func() == 'part.tasks.notify_low_stock_if_required'
+        ]
+        self.assertEqual(
+            {task.args()[0] for task in low_stock_tasks},
+            {component.pk for component in components},
+        )
+
+        # All lines should be fully consumed, and no allocations should remain
+        self.assertEqual(build.allocated_stock.count(), 0)
+        self.assertEqual(build.build_lines.count(), N)
+
+        for line in build.build_lines.all():
+            self.assertEqual(line.consumed, required_quantity)
+
+        # Check that each original stock item was correctly reduced / split
+        for idx, (component, items) in enumerate(
+            zip(components, stock_items, strict=True)
+        ):
+            for si in items:
+                si.refresh_from_db()
+
+            if idx % MULTI_ITEM_STRIDE == 0:
+                # Two different stock items were allocated against this line - each
+                # should be fully consumed directly (no splitting), and combined they
+                # should cover the full required quantity
+                self.assertEqual(len(items), 2)
+
+                consumed_total = sum(si.quantity for si in items)
+                self.assertEqual(consumed_total, required_quantity)
+
+                for si in items:
+                    self.assertEqual(si.consumed_by, build)
+
+                    self.assertFalse(
+                        StockItemTracking.objects.filter(
+                            item=si,
+                            tracking_type=StockHistoryCode.SPLIT_CHILD_ITEM.value,
+                        ).exists()
+                    )
+                    self.assertTrue(
+                        StockItemTracking.objects.filter(
+                            item=si,
+                            tracking_type=StockHistoryCode.BUILD_CONSUMED.value,
+                            deltas__buildorder=build.pk,
+                        ).exists()
+                    )
+
+                # Total quantity of stock for this component is unchanged
+                total_quantity = StockItem.objects.filter(part=component).aggregate(
+                    total=Sum('quantity')
+                )['total']
+                self.assertEqual(total_quantity, required_quantity)
+
+                continue
+
+            stock_item = items[0]
+
+            if idx % 2:
+                # Excess stock - original item should be split, retaining the remainder
+                self.assertEqual(stock_item.quantity, 25)
+                self.assertIsNone(stock_item.consumed_by)
+
+                # A new stock item should have been split off, and consumed by the build
+                consumed_items = StockItem.objects.filter(
+                    part=component, consumed_by=build
+                )
+                self.assertEqual(consumed_items.count(), 1)
+                consumed_item = consumed_items.first()
+                self.assertEqual(consumed_item.quantity, required_quantity)
+                self.assertNotEqual(consumed_item.pk, stock_item.pk)
+
+                # Split tracking entries: one on the new (split-off) item, one on the original
+                self.assertTrue(
+                    StockItemTracking.objects.filter(
+                        item=consumed_item,
+                        tracking_type=StockHistoryCode.SPLIT_FROM_PARENT.value,
+                        deltas__stockitem=stock_item.pk,
+                    ).exists()
+                )
+                self.assertTrue(
+                    StockItemTracking.objects.filter(
+                        item=stock_item,
+                        tracking_type=StockHistoryCode.SPLIT_CHILD_ITEM.value,
+                    ).exists()
+                )
+
+                # The ITEM_SPLIT event should reference the new item as 'id' and the
+                # original item as 'parent'
+                self.assertTrue(
+                    any(
+                        task.kwargs()
+                        == {'id': consumed_item.pk, 'parent': stock_item.pk}
+                        for task in split_event_tasks
+                    )
+                )
+
+                # Consumption is recorded against the new (split-off) item
+                self.assertTrue(
+                    StockItemTracking.objects.filter(
+                        item=consumed_item,
+                        tracking_type=StockHistoryCode.BUILD_CONSUMED.value,
+                        deltas__buildorder=build.pk,
+                    ).exists()
+                )
+            else:
+                # Exact stock - original item should be consumed directly (no split)
+                self.assertEqual(stock_item.quantity, required_quantity)
+                self.assertEqual(stock_item.consumed_by, build)
+
+                # No split occurred - consumption is recorded directly against this item
+                self.assertFalse(
+                    StockItemTracking.objects.filter(
+                        item=stock_item,
+                        tracking_type=StockHistoryCode.SPLIT_CHILD_ITEM.value,
+                    ).exists()
+                )
+                self.assertTrue(
+                    StockItemTracking.objects.filter(
+                        item=stock_item,
+                        tracking_type=StockHistoryCode.BUILD_CONSUMED.value,
+                        deltas__buildorder=build.pk,
+                    ).exists()
+                )
+
+            # In either case, total quantity of stock for this component is unchanged
+            total_quantity = StockItem.objects.filter(part=component).aggregate(
+                total=Sum('quantity')
+            )['total']
+            self.assertEqual(total_quantity, required_quantity + (25 if idx % 2 else 0))
+
+        # The total number of StockItem instances in the DB has also increased
+        self.assertEqual(StockItem.objects.count(), N_STOCK_ITEMS + 1 + (N // 2))
+
+    def test_consume_tracked_allocations(self):
+        """Test consuming of tracked allocations against a BuildOrder."""
+        set_global_setting('ENABLE_PLUGINS_EVENTS', True, change_user=None)
+
+        tracked_assembly = Part.objects.create(
+            name='Tracked Test Assembly',
+            description='Trackable assembly for install-path testing',
+            assembly=True,
+            trackable=True,
+        )
+
+        tracked_component = Part.objects.create(
+            name='Tracked Test Component',
+            description='Trackable component for install-path testing',
+            trackable=True,
+            component=True,
+        )
+
+        BomItem.objects.create(
+            part=tracked_assembly, sub_part=tracked_component, quantity=1
+        )
+
+        tracked_build = Build.objects.create(
+            part=tracked_assembly,
+            reference='BO-99002',
+            quantity=2,
+            title='Tracked Test Build',
+        )
+
+        serials = ['901', '902']
+
+        for sn in serials:
+            StockItem.objects.create(part=tracked_component, quantity=1, serial=sn)
+
+        # Auto-allocate serialized outputs against matching component serial numbers
+        response = self.post(
+            reverse('api-build-output-create', kwargs={'pk': tracked_build.pk}),
+            {
+                'quantity': 2,
+                'serial_numbers': ', '.join(serials),
+                'auto_allocate': True,
+            },
+            expected_code=201,
+        )
+
+        outputs = {
+            entry['serial']: StockItem.objects.get(
+                part=tracked_assembly, serial=entry['serial']
+            )
+            for entry in response.data
+        }
+        tracked_components = {
+            sn: StockItem.objects.get(part=tracked_component, serial=sn)
+            for sn in serials
+        }
+
+        OrmQ.objects.all().delete()
+
+        # Consume the tracked allocations directly (rather than completing the output),
+        # to route them through Build.complete_allocations() and exercise the install path
+        with self.settings(
+            PLUGIN_TESTING_EVENTS=True, PLUGIN_TESTING_EVENTS_ASYNC=True
+        ):
+            self.post(
+                reverse('api-build-consume', kwargs={'pk': tracked_build.pk}),
+                {
+                    'lines': [
+                        {'build_line': line.pk}
+                        for line in tracked_build.build_lines.all()
+                    ]
+                },
+                expected_code=200,
+            )
+
+        self.assertEqual(tracked_build.allocated_stock.count(), 0)
+
+        install_event_tasks = [
+            task
+            for task in OrmQ.objects.all()
+            if task.func() == 'plugin.base.event.events.register_event'
+            and task.args() == (StockEvents.ITEM_INSTALLED_INTO_ASSEMBLY,)
+        ]
+        self.assertEqual(len(install_event_tasks), len(serials))
+
+        for sn in serials:
+            component_item = tracked_components[sn]
+            output_item = outputs[sn]
+
+            component_item.refresh_from_db()
+
+            # The component is now installed into (and consumed by) the output
+            self.assertEqual(component_item.belongs_to, output_item)
+            self.assertEqual(component_item.consumed_by, tracked_build)
+
+            self.assertTrue(
+                any(
+                    task.kwargs()
+                    == {'id': component_item.pk, 'assembly_id': output_item.pk}
+                    for task in install_event_tasks
+                )
+            )
+
+            self.assertTrue(
+                StockItemTracking.objects.filter(
+                    item=component_item,
+                    tracking_type=StockHistoryCode.INSTALLED_INTO_ASSEMBLY.value,
+                    deltas__stockitem=output_item.pk,
+                ).exists()
+            )
+            self.assertTrue(
+                StockItemTracking.objects.filter(
+                    item=output_item,
+                    tracking_type=StockHistoryCode.INSTALLED_CHILD_ITEM.value,
+                    deltas__stockitem=component_item.pk,
+                ).exists()
+            )
+
 
 class BuildCustomStatusTest(BuildAPITest):
     """Tests for custom status values on Build orders."""
@@ -2396,3 +2825,149 @@ class BuildAutoAllocateAPITest(InvenTreeAPITestCase):
 
         self.assertTrue(alloc_a.exists())
         self.assertTrue(alloc_b.exists())
+
+    # ------------------------------------------------------------------
+    # Variant stock / sort priority
+    # ------------------------------------------------------------------
+
+    def test_direct_match_preferred_over_variant_stock(self):
+        """Direct part matches are allocated ahead of variant stock (sort priority 1 vs 2)."""
+        build = self._make_build(quantity=5)
+
+        bom_item = BomItem.objects.get(part=self.assembly, sub_part=self.component)
+        bom_item.allow_variants = True
+        bom_item.save()
+
+        self.component.is_template = True
+        self.component.save()
+
+        variant = Part.objects.create(
+            name='AutoAlloc Component Variant',
+            description='',
+            component=True,
+            variant_of=self.component,
+        )
+
+        direct_stock = StockItem.objects.create(part=self.component, quantity=5)
+        StockItem.objects.create(part=variant, quantity=100)
+
+        self.post(self._url(build.pk), {'interchangeable': True}, expected_code=200)
+
+        allocs = BuildItem.objects.filter(build_line__build=build)
+        self.assertEqual(allocs.count(), 1)
+        self.assertEqual(allocs.first().stock_item, direct_stock)
+
+    # ------------------------------------------------------------------
+    # Location filtering
+    # ------------------------------------------------------------------
+
+    def test_location_filters_allocation(self):
+        """The 'location' filter restricts allocation to stock within that location."""
+        build = self._make_build(quantity=5)
+
+        in_location = StockItem.objects.create(
+            part=self.component, quantity=5, location=self.loc_a
+        )
+        StockItem.objects.create(part=self.component, quantity=100, location=self.loc_b)
+
+        self.post(
+            self._url(build.pk),
+            {'location': self.loc_a.pk, 'interchangeable': True},
+            expected_code=200,
+        )
+
+        allocs = BuildItem.objects.filter(build_line__build=build)
+        self.assertEqual(allocs.count(), 1)
+        self.assertEqual(allocs.first().stock_item, in_location)
+
+    def test_exclude_location_filters_allocation(self):
+        """The 'exclude_location' filter excludes stock within that location."""
+        build = self._make_build(quantity=5)
+
+        StockItem.objects.create(part=self.component, quantity=100, location=self.loc_a)
+        allowed = StockItem.objects.create(
+            part=self.component, quantity=5, location=self.loc_b
+        )
+
+        self.post(
+            self._url(build.pk),
+            {'exclude_location': self.loc_a.pk, 'interchangeable': True},
+            expected_code=200,
+        )
+
+        allocs = BuildItem.objects.filter(build_line__build=build)
+        self.assertEqual(allocs.count(), 1)
+        self.assertEqual(allocs.first().stock_item, allowed)
+
+    # ------------------------------------------------------------------
+    # Large number of build lines (performance / bulk allocation)
+    # ------------------------------------------------------------------
+
+    def test_many_build_lines_allocated_correctly(self):
+        """A build with 100 BOM lines, each with sufficient dedicated stock, is fully auto-allocated.
+
+        This test is a performance benchmark for the auto-allocate endpoint,
+        to ensure that auto-allocating 100x items does not cause an excessive amount of DB hits.
+        """
+        assembly = Part.objects.create(
+            name='AutoAlloc Big Assembly', description='', assembly=True
+        )
+        build = Build.objects.create(
+            part=assembly,
+            reference=f'BO-{9000 + Build.objects.count():04d}',
+            quantity=10,
+        )
+
+        components = [
+            Part.objects.create(
+                name=f'AutoAlloc Bulk Component {i}',
+                description='',
+                component=True,
+                tree_id=0,
+                level=0,
+                lft=0,
+                rght=0,
+            )
+            for i in range(100)
+        ]
+
+        # Create BomItem for each component, and a stock item with sufficient quantity to cover the build
+        bom_items = [
+            BomItem(part=assembly, sub_part=component, quantity=13)
+            for component in components
+        ]
+
+        BomItem.objects.bulk_create(bom_items)
+
+        # Create multiple stock items for each component, with a total sufficient stock to fulfil each line
+        stock_items = []
+
+        for _i in range(4):
+            for component in components:
+                # Total of 160 stock for each component
+                stock_items.append(StockItem(part=component, quantity=40))
+
+        StockItem.objects.bulk_create(stock_items)
+
+        build.create_build_line_items()
+
+        self.assertEqual(build.build_lines.count(), 100)
+
+        self.post(
+            self._url(build.pk),
+            {'interchangeable': True},
+            expected_code=200,
+            benchmark=True,
+            max_query_count=150,
+        )
+
+        allocs = BuildItem.objects.filter(build_line__build=build)
+
+        # Every line should have received exactly one allocation, covering the full requirement.
+        self.assertEqual(allocs.count(), 400)
+
+        for component in components:
+            fa = allocs.filter(stock_item__part=component)
+            self.assertEqual(fa.count(), 4)
+            allocated = sum(a.quantity for a in fa)
+            self.assertEqual(allocated, 130)  # 130 allocated to each line
diff --git a/src/backend/InvenTree/build/test_build.py b/src/backend/InvenTree/build/test_build.py
index 8a8cd7fce4..a5f50aa51d 100644
--- a/src/backend/InvenTree/build/test_build.py
+++ b/src/backend/InvenTree/build/test_build.py
@@ -483,6 +483,8 @@ class BuildTest(BuildTestBase):
 
         self.build.complete_build(None)
 
+        # The status is updated by the (synchronous, in tests) background task
+        self.build.refresh_from_db()
         self.assertEqual(self.build.status, status.BuildStatus.COMPLETE)
 
         # Check stock items are in expected state.
@@ -664,6 +666,8 @@ class BuildTest(BuildTestBase):
 
         self.build.complete_build(None)
 
+        # The status is updated by the (synchronous, in tests) background task
+        self.build.refresh_from_db()
         self.assertEqual(self.build.status, status.BuildStatus.COMPLETE)
 
         # the original BuildItem objects should have been deleted!
@@ -724,37 +728,50 @@ class BuildTest(BuildTestBase):
         self.build.refresh_from_db()
         self.assertEqual(self.build.completed, 10)
 
-    def test_complete_allocation_stale_build_line(self):
-        """The 'consumed' count is incremented atomically at the database level.
+    def test_complete_allocations_sums_consumed(self):
+        """Completing multiple allocations against one BuildLine sums the consumed count.
 
-        Simulates two concurrent workers completing different allocations
-        against the same BuildLine, each holding its own (stale) copy of the line.
+        (Concurrent completions are serialized by the Build row lock inside
+        complete_allocations, so only the sequential arithmetic is tested here.)
         """
         self.build.issue_build()
 
         self.allocate_stock(None, {self.stock_1_1: 3, self.stock_1_2: 5})
 
-        alloc_a, alloc_b = BuildItem.objects.filter(build_line=self.line_1).order_by(
-            'pk'
+        self.build.complete_allocations(
+            BuildItem.objects.filter(build_line=self.line_1), user=self.user
         )
 
-        # Cache a separate copy of the BuildLine on each allocation
-        self.assertEqual(alloc_a.build_line.consumed, 0)
-        self.assertEqual(alloc_b.build_line.consumed, 0)
-
-        alloc_a.complete_allocation(user=self.user)
-        alloc_b.complete_allocation(user=self.user)
-
         # Both consumed quantities must be counted
         self.line_1.refresh_from_db()
         self.assertEqual(self.line_1.consumed, 8)
 
+    def test_complete_allocation_wrapper(self):
+        """BuildItem.complete_allocation() is a thin wrapper around Build.complete_allocations().
+
+        Regression test: complete_allocation() was removed when complete_allocations() was
+        introduced, but some call sites still complete a single BuildItem at a time.
+        """
+        self.build.issue_build()
+
+        self.allocate_stock(None, {self.stock_1_1: 3})
+        alloc = BuildItem.objects.get(build_line=self.line_1, stock_item=self.stock_1_1)
+
+        alloc.complete_allocation(user=self.user)
+
+        self.stock_1_1.refresh_from_db()
+        self.line_1.refresh_from_db()
+
+        self.assertEqual(self.stock_1_1.consumed_by, self.build)
+        self.assertEqual(self.line_1.consumed, 3)
+        self.assertFalse(BuildItem.objects.filter(pk=alloc.pk).exists())
+
     def test_complete_zero_quantity_allocation(self):
-        """A zero-quantity allocation is removed cleanly on completion.
+        """A zero-quantity allocation is skipped cleanly on completion.
 
         Regression test: completing a BuildItem with quantity=0 (permitted by
-        the model validators) crashed with an AttributeError, blocking build
-        completion until the empty allocation was manually removed.
+        the model validators) crashed, blocking build completion until the
+        empty allocation was manually removed.
         """
         self.build.issue_build()
 
@@ -765,10 +782,11 @@ class BuildTest(BuildTestBase):
 
         n_items = StockItem.objects.count()
 
-        alloc.complete_allocation(user=self.user)
+        # Completing the allocation must not crash, and performs no stock operations
+        self.build.complete_allocations(
+            BuildItem.objects.filter(pk=alloc.pk), user=self.user
+        )
 
-        # The empty allocation is deleted, with no stock operations performed
-        self.assertFalse(BuildItem.objects.filter(pk=alloc.pk).exists())
         self.assertEqual(StockItem.objects.count(), n_items)
 
         self.stock_1_2.refresh_from_db()
@@ -778,8 +796,10 @@ class BuildTest(BuildTestBase):
         self.line_1.refresh_from_db()
         self.assertEqual(self.line_1.consumed, 0)
 
+        alloc.delete()
+
         # An allocation whose stock item has been depleted elsewhere
-        # is also removed cleanly (allocated quantity clamps to zero)
+        # is also skipped cleanly (allocated quantity clamps to zero)
         depleted = StockItem.objects.create(
             part=self.sub_part_1, quantity=5, delete_on_deplete=False
         )
@@ -791,9 +811,9 @@ class BuildTest(BuildTestBase):
         depleted.refresh_from_db()
         self.assertEqual(depleted.quantity, 0)
 
-        alloc.complete_allocation(user=self.user)
-
-        self.assertFalse(BuildItem.objects.filter(pk=alloc.pk).exists())
+        self.build.complete_allocations(
+            BuildItem.objects.filter(pk=alloc.pk), user=self.user
+        )
 
         depleted.refresh_from_db()
         self.assertIsNone(depleted.consumed_by)
@@ -1159,11 +1179,8 @@ class AutoAllocationTests(BuildTestBase):
         self.assertEqual(self.line_1.allocations.count(), 2)
         self.assertEqual(self.line_2.allocations.count(), 6)
 
-        for item in self.line_1.allocations.all():
-            item.complete_allocation()
-
-        for item in self.line_2.allocations.all():
-            item.complete_allocation()
+        self.build.complete_allocations(self.line_1.allocations.all())
+        self.build.complete_allocations(self.line_2.allocations.all())
 
         self.line_1.refresh_from_db()
         self.line_2.refresh_from_db()
@@ -1360,6 +1377,9 @@ class ExternalBuildTest(InvenTreeAPITestCase):
 
         # Mark the build order as completed
         build.complete_build(self.user)
+
+        # The status is updated by the (synchronous, in tests) background task
+        build.refresh_from_db()
         self.assertEqual(build.status, BuildStatus.COMPLETE)
 
         # Receive the rest of the line item
@@ -1461,6 +1481,68 @@ class BuildTaskTests(BuildTestBase):
         self.build.complete_build_output(self.output_1, None)
         self.build.complete_build_output(self.output_2, None)
 
+    def test_complete_build_task_is_idempotent(self):
+        """A duplicated completion task run must be a no-op.
+
+        Regression test: the completion task had no build lock and no status
+        re-check, so a redelivered (or double-enqueued) task re-ran the
+        completion side effects (duplicate COMPLETED event and notifications).
+        """
+        from build.tasks import complete_build
+
+        self._setup_complete_build()
+        self.build.complete_build(self.user)
+
+        self.build.refresh_from_db()
+        self.assertEqual(self.build.status, BuildStatus.COMPLETE)
+
+        n_consumed = StockItem.objects.filter(consumed_by=self.build).count()
+        n_tracking = StockItemTracking.objects.count()
+
+        # A redelivered task run must skip based on the (locked) database state
+        with mock.patch('build.tasks.trigger_event') as trigger:
+            complete_build(self.build.pk, self.user.pk)
+            trigger.assert_not_called()
+
+        self.assertEqual(
+            StockItem.objects.filter(consumed_by=self.build).count(), n_consumed
+        )
+        self.assertEqual(StockItemTracking.objects.count(), n_tracking)
+
+    def test_allocate_stock_merges_quantities(self):
+        """Repeated allocations against the same (line, stock item) accumulate.
+
+        Regression test: allocate_stock() never added merged BuildItems to its
+        'to_update' set, so allocating against an existing allocation silently
+        discarded the requested quantity. Duplicate entries within a single
+        request also overwrote (rather than summed) each other.
+        """
+        self.build.issue_build()
+
+        items = [
+            {'build_line': self.line_1, 'stock_item': self.stock_1_2, 'quantity': 10}
+        ]
+
+        self.build.allocate_stock(items)
+
+        alloc = BuildItem.objects.get(build_line=self.line_1, stock_item=self.stock_1_2)
+        self.assertEqual(alloc.quantity, 10)
+
+        # A second allocation against the same (line, stock item) merges quantities
+        self.build.allocate_stock(items)
+
+        alloc.refresh_from_db()
+        self.assertEqual(alloc.quantity, 20)
+
+        # Duplicate entries within a single request are also merged
+        self.build.allocate_stock([
+            {'build_line': self.line_1, 'stock_item': self.stock_1_1, 'quantity': 1},
+            {'build_line': self.line_1, 'stock_item': self.stock_1_1, 'quantity': 2},
+        ])
+
+        alloc = BuildItem.objects.get(build_line=self.line_1, stock_item=self.stock_1_1)
+        self.assertEqual(alloc.quantity, 3)
+
     # -----------------------------------------------------------------------
     # complete_build_outputs / scrap_build_outputs tasks
     # -----------------------------------------------------------------------
@@ -1877,7 +1959,7 @@ class BuildTrimAllocatedStockConcurrencyTest(TransactionTestCase):
 
 @skipUnlessDBFeature('has_select_for_update')
 class BuildSubtractAllocatedStockConcurrencyTest(TransactionTestCase):
-    """Genuine cross-transaction regression test for Build.subtract_allocated_stock().
+    """Genuine cross-transaction regression test for Build.complete_outstanding_allocations().
 
     Uses two real threads (each with its own database connection) to reproduce
     duplicated/overlapping execution - e.g. a redelivered 'complete_build' or
@@ -1923,7 +2005,7 @@ class BuildSubtractAllocatedStockConcurrencyTest(TransactionTestCase):
         )
 
     def test_subtract_allocated_stock_is_not_processed_twice(self):
-        """A duplicated/concurrent call to subtract_allocated_stock() must not double-consume.
+        """A duplicated/concurrent call to complete_outstanding_allocations() must not double-consume.
 
         Regression test: BuildItem allocation rows were read via a plain (unlocked)
         queryset before being consumed and deleted. Two overlapping calls to this
@@ -1939,7 +2021,7 @@ class BuildSubtractAllocatedStockConcurrencyTest(TransactionTestCase):
         def run_subtract():
             try:
                 start_barrier.wait()
-                self.build.subtract_allocated_stock(self.user)
+                self.build.complete_outstanding_allocations(self.user)
             except Exception as exc:  # pragma: no cover - surfaced via errors list
                 errors.append(exc)
             finally:
diff --git a/src/backend/InvenTree/part/models.py b/src/backend/InvenTree/part/models.py
index 1db047143a..aa4c1f2666 100644
--- a/src/backend/InvenTree/part/models.py
+++ b/src/backend/InvenTree/part/models.py
@@ -3694,6 +3694,7 @@ class BomItem(InvenTree.models.MetadataMixin, InvenTree.models.InvenTreeModel):
         allow_variants: bool = True,
         allow_substitutes: bool = True,
         allow_inactive: bool = True,
+        variant_parts=None,
     ):
         """Return a list of valid parts which can be allocated against this BomItem.
 
@@ -3701,6 +3702,8 @@ class BomItem(InvenTree.models.MetadataMixin, InvenTree.models.InvenTreeModel):
             allow_variants: If True, include variants of the sub_part
             allow_substitutes: If True, include any directly specified substitute parts
             allow_inactive: If True, include inactive parts in the returned list
+            variant_parts: Optional pre-fetched iterable of sub_part's descendants,
+                to avoid re-querying the part tree when the caller already has it
 
         Includes:
         - The referenced sub_part
@@ -3714,7 +3717,9 @@ class BomItem(InvenTree.models.MetadataMixin, InvenTree.models.InvenTreeModel):
 
         # Variant parts (if allowed)
         if allow_variants and self.allow_variants:
-            for variant in self.sub_part.get_descendants(include_self=False):
+            if variant_parts is None:
+                variant_parts = self.sub_part.get_descendants(include_self=False)
+            for variant in variant_parts:
                 parts.add(variant)
 
         # Substitute parts
diff --git a/src/backend/InvenTree/plugin/events.py b/src/backend/InvenTree/plugin/events.py
index 2097767440..a36bfb97e8 100644
--- a/src/backend/InvenTree/plugin/events.py
+++ b/src/backend/InvenTree/plugin/events.py
@@ -2,6 +2,7 @@
 
 from generic.events import BaseEventEnum
 from plugin.base.event.events import (
+    batch_events,
     bulk_trigger_event,
     process_event,
     register_event,
@@ -18,6 +19,7 @@ class PluginEvents(BaseEventEnum):
 
 __all__ = [
     'PluginEvents',
+    'batch_events',
     'bulk_trigger_event',
     'process_event',
     'register_event',
diff --git a/src/backend/InvenTree/stock/models.py b/src/backend/InvenTree/stock/models.py
index 1358bc160e..2d0a663014 100644
--- a/src/backend/InvenTree/stock/models.py
+++ b/src/backend/InvenTree/stock/models.py
@@ -1670,6 +1670,53 @@ class StockItem(
         """Return the quantity of this StockItem which is *not* allocated."""
         return max(self.quantity - self.allocation_count(), 0)
 
+    @staticmethod
+    def bulk_allocation_count(stock_items) -> dict:
+        """Bulk-compute the total allocated quantity (builds + sales orders + transfer orders) for a set of StockItem objects.
+
+        Mirrors the per-instance calculation in `allocation_count()`, but resolves the whole
+        set in 3 aggregate queries (one per allocation type) rather than one query per
+        allocation type *per item* - critical for validating requests which reference many
+        stock items at once.
+
+        Returns:
+            A {stock_item.pk: total_allocated_quantity} dict. Items with no allocations
+            of any kind are omitted (i.e. callers should default missing pks to zero).
+        """
+        pks = [item.pk for item in stock_items]
+
+        totals = {}
+
+        def _accumulate(queryset, group_field):
+            rows = queryset.values(group_field).annotate(
+                total=Coalesce(Sum('quantity'), Decimal(0))
+            )
+            for row in rows:
+                pk = row[group_field]
+                totals[pk] = totals.get(pk, Decimal(0)) + row['total']
+
+        _accumulate(
+            build.models.BuildItem.objects.filter(stock_item__in=pks), 'stock_item'
+        )
+
+        _accumulate(
+            order.models.SalesOrderAllocation.objects.filter(
+                item__in=pks,
+                line__order__status__in=SalesOrderStatusGroups.OPEN,
+                shipment__shipment_date=None,
+            ),
+            'item',
+        )
+
+        _accumulate(
+            order.models.TransferOrderAllocation.objects.filter(
+                item__in=pks, line__order__status__in=TransferOrderStatusGroups.OPEN
+            ),
+            'item',
+        )
+
+        return totals
+
     def can_delete(self):
         """Can this stock item be deleted?
 
@@ -2817,6 +2864,7 @@ class StockItem(
         linking the resulting tracking entry back to the order which triggered it.
         """
         return [
+            'stockitem',
             'stockitemlocation',
             'transferorder',
             'purchaseorder',
diff --git a/src/backend/InvenTree/stock/tests.py b/src/backend/InvenTree/stock/tests.py
index 810522e8b1..edc2b620c3 100644
--- a/src/backend/InvenTree/stock/tests.py
+++ b/src/backend/InvenTree/stock/tests.py
@@ -351,6 +351,18 @@ class StockTest(StockTestBase):
         stock.splitStock(stock.quantity, None, self.user)
         self.assertEqual(StockItem.objects.filter(part=3).count(), n + 1)
 
+    def test_split_stock_tracking_deltas(self):
+        """The parent's SPLIT_CHILD_ITEM tracking entry must reference the new child item."""
+        parent = StockItem.objects.get(id=1234)
+        child = parent.splitStock(100, None, self.user)
+
+        parent_entry = parent.tracking_info.filter(
+            tracking_type=StockHistoryCode.SPLIT_CHILD_ITEM.value
+        ).first()
+
+        self.assertIsNotNone(parent_entry)
+        self.assertEqual(parent_entry.deltas.get('stockitem'), child.pk)
+
     def test_delete_reparents_children(self):
         """Test that deleting an intermediate item re-links children to the grandparent."""
         grandparent = StockItem.objects.get(id=1234)
diff --git a/src/frontend/tests/pages/pui_build.spec.ts b/src/frontend/tests/pages/pui_build.spec.ts
index 1ba14e45b6..937d187a6c 100644
--- a/src/frontend/tests/pages/pui_build.spec.ts
+++ b/src/frontend/tests/pages/pui_build.spec.ts
@@ -627,6 +627,16 @@ test('Build Order - Consume Stock', async ({ browser }) => {
 
   await page.getByText('Fully consumed').first().waitFor();
   await page.getByText('15 / 15').first().waitFor();
+
+  // There should now not be any allocated stock remaining
+  await loadTab(page, 'Allocated Stock');
+  await page.getByText('No records found').first().waitFor();
+
+  await loadTab(page, 'Consumed Stock');
+  await page.getByRole('cell', { name: 'Thumbnail C_1uF_0805' }).waitFor();
+  await page.getByRole('cell', { name: 'Thumbnail M3x8 Torx' }).waitFor();
+  await page.getByRole('cell', { name: 'Thumbnail R_10K_0805_1%' }).waitFor();
+  await page.getByText('1 - 3 / 3').waitFor();
 });
 
 test('Build Order - Tracked Outputs', async ({ browser }) => {