Enforce the intake allocation limit in the lineage API

The ops lineage screen refuses to allocate more of an intake than was
received, counting every batch, but POST /api/intake-batch-lineage/ accepted
9999 kg against a 100 kg intake. Move the rule into a shared helper so the
screen and the serializer refuse the same allocations.

apps/compliance/serializers.py

@@ -1,5 +1,6 @@
 from rest_framework import serializers
 
+from . import services
 from .models import (
     ProductSpecification, SpecParameter,
     Facility, OriginZone, CollectorGroup, HarvestIntake, IntakeBatchLineage,
@@ -68,6 +69,21 @@ class IntakeBatchLineageSerializer(serializers.ModelSerializer):
         fields = "__all__"
         read_only_fields = ["id", "created_at"]
 
+    def validate(self, attrs):
+        intake = attrs.get("harvest_intake") or getattr(
+            self.instance, "harvest_intake", None
+        )
+        weight = attrs.get("weight_allocated_kg")
+        if weight is None and self.instance is not None:
+            weight = self.instance.weight_allocated_kg
+        if intake is not None and weight is not None:
+            message = services.intake_allocation_error(
+                intake, weight, exclude_lineage=self.instance
+            )
+            if message:
+                raise serializers.ValidationError({"weight_allocated_kg": message})
+        return attrs
+
 
 class EUDREvidenceSerializer(serializers.ModelSerializer):
     verified_by_name = serializers.CharField(source="verified_by.full_name", read_only=True)

apps/compliance/services.py

@@ -1,6 +1,7 @@
 from decimal import Decimal
 
 from django.db import transaction
+from django.db.models import Sum
 from django.utils import timezone
 
 from apps.batches.models import ProcessingBatch
@@ -177,6 +178,43 @@ def ensure_due_diligence_case(batch):
     return case
 
 
+def intake_allocation_error(intake, weight, *, exclude_batch=None, exclude_lineage=None):
+    """Return a message when ``weight`` cannot be allocated from ``intake``.
+
+    An intake is a finite quantity of shea, so the rule counts **every** batch
+    that has already drawn from it, not just the one being saved. ``exclude_batch``
+    and ``exclude_lineage`` keep the caller's own allocation out of that total.
+    Shared by the ops lineage formset and the lineage API serializer so the screen
+    and the API refuse the same allocations.
+    """
+    from apps.compliance.models import IntakeBatchLineage
+
+    received = intake.weight_kg
+    if received is None:
+        # Records that arrived through the API can be missing their weight.
+        # Nothing to check against, so nothing is claimed.
+        return None
+    if weight > received:
+        return (
+            f"{intake.intake_reference} received {received} kg; "
+            f"{weight} kg cannot be allocated from it."
+        )
+    elsewhere_qs = IntakeBatchLineage.objects.filter(harvest_intake=intake)
+    if exclude_batch is not None:
+        elsewhere_qs = elsewhere_qs.exclude(batch=exclude_batch)
+    if exclude_lineage is not None:
+        elsewhere_qs = elsewhere_qs.exclude(pk=exclude_lineage.pk)
+    elsewhere = elsewhere_qs.aggregate(total=Sum("weight_allocated_kg"))["total"] or Decimal("0")
+    if elsewhere + weight > received:
+        others = ", ".join(sorted({row.batch.batch_reference for row in elsewhere_qs}))
+        return (
+            f"{intake.intake_reference} received {received} kg and "
+            f"{elsewhere} kg of it is already allocated to {others}. "
+            f"Only {received - elsewhere} kg is left."
+        )
+    return None
+
+
 def _lineage(batch):
     return list(batch.intake_lineage.select_related("harvest_intake", "harvest_intake__origin_zone", "harvest_intake__collector_group").all())
 

apps/compliance/tests.py

@@ -240,6 +240,65 @@ class EUDRWorkflowTests(TestCase):
         self.assertEqual(case.statement_references.count(), 1)
 
 
+class LineageAllocationAPITests(TestCase):
+    """The API refuses the same over-allocation the ops lineage screen refuses."""
+
+    def setUp(self):
+        self.ops = make_user("lineage.ops@ixg.local", "ops")
+        self.batch = make_batch(self.ops)
+        facility = Facility.objects.create(
+            name="Lineage Facility", country="Nigeria", region="Katsina"
+        )
+        zone = OriginZone.objects.create(
+            facility=facility, zone_code="LIN-01", country="Nigeria",
+            region="Katsina", verification_status=OriginZone.VerificationStatus.VERIFIED,
+        )
+        group = CollectorGroup.objects.create(name="Lineage Group", origin_zone=zone)
+        self.intake = HarvestIntake.objects.create(
+            intake_reference="INT-LIN-001", collector_group=group,
+            origin_zone=zone, intake_date="2026-01-10",
+            weight_kg=Decimal("100.00"),
+        )
+        self.client = APIClient()
+        self.client.force_authenticate(user=self.ops)
+
+    def post(self, weight):
+        return self.client.post(
+            "/api/intake-batch-lineage/",
+            {
+                "batch": str(self.batch.id),
+                "harvest_intake": str(self.intake.id),
+                "weight_allocated_kg": weight,
+            },
+            format="json",
+        )
+
+    def test_an_allocation_larger_than_the_intake_is_refused(self):
+        response = self.post("9999.00")
+        self.assertEqual(response.status_code, 400)
+        self.assertFalse(IntakeBatchLineage.objects.exists())
+
+    def test_an_allocation_that_overdraws_with_another_batch_is_refused(self):
+        other = make_batch(self.ops)
+        IntakeBatchLineage.objects.create(
+            batch=other, harvest_intake=self.intake,
+            weight_allocated_kg=Decimal("80.00"),
+        )
+        response = self.post("30.00")
+        self.assertEqual(response.status_code, 400)
+        self.assertEqual(IntakeBatchLineage.objects.count(), 1)
+
+    def test_what_is_left_of_the_intake_can_be_allocated(self):
+        other = make_batch(self.ops)
+        IntakeBatchLineage.objects.create(
+            batch=other, harvest_intake=self.intake,
+            weight_allocated_kg=Decimal("80.00"),
+        )
+        response = self.post("20.00")
+        self.assertEqual(response.status_code, 201)
+        self.assertEqual(IntakeBatchLineage.objects.count(), 2)
+
+
 class BuyerIsolationTests(TestCase):
     def setUp(self):
         self.buyer = make_user("iso.buyer@ixg.local", "buyer")