From 4b4477d35921e1098913a398b8adbbe58b99cd88 Mon Sep 17 00:00:00 2001
From: ixgadmin <admin@ixg.local>
Date: Thu, 17 Sep 2026 14:48:06 +0000
Subject: [PATCH] Add batch create, edit and lineage screens to the ops portal

The batch form refuses to type over a QC status or move a failed batch, and
the lineage formset refuses an allocation the intake cannot support. The
origin-facility help text and the QC status card now describe what the form
actually does, and buyer POSTs to the create and edit screens are refused.
---
 apps/ops/forms.py                | 376 +++++++++++++++++++++++++-
 apps/ops/tests.py                | 446 ++++++++++++++++++++++++++++++-
 apps/ops/urls.py                 |   3 +
 apps/ops/views.py                | 170 +++++++++++-
 templates/ops/batch_form.html    |  92 +++++++
 templates/ops/batch_lineage.html | 114 ++++++++
 templates/ops/batch_list.html    |   5 +-
 7 files changed, 1186 insertions(+), 20 deletions(-)
 create mode 100644 templates/ops/batch_form.html
 create mode 100644 templates/ops/batch_lineage.html

diff --git a/apps/ops/forms.py b/apps/ops/forms.py
index 4d2b101..eed8ada 100644
--- a/apps/ops/forms.py
+++ b/apps/ops/forms.py
@@ -1,19 +1,38 @@
 """Forms for the ops portal.
 
-The intake form is deliberately **stricter than the model**. ``HarvestIntake``
-allows almost everything to be blank, because records also arrive through the
-API, and the platform copes by flagging the gaps later: the EUDR completeness
-check has an "intake data" test whose whole job is to catch a missing date or
-weight. A portal form that let the same blanks through would simply move that
-problem to a screen where a person could have prevented it.
+Both forms here share a principle: **stricter than the model.**
 
-So every field the checks read is required here — reference, collector group,
-origin zone, date, weight and grade — while the fields that are genuinely
-optional at intake (receiving officer, the supporting file's name) stay optional.
+The intake form is stricter about *completeness*. ``HarvestIntake`` allows almost
+everything to be blank, because records also arrive through the API, and the
+platform copes by flagging the gaps later: the EUDR completeness check has an
+"intake data" test whose whole job is to catch a missing date or weight. A portal
+form that let the same blanks through would simply move that problem to a screen
+where a person could have prevented it. So every field the checks read is
+required — reference, collector group, origin zone, date, weight and grade — and
+the fields that are genuinely optional at intake stay optional.
+
+The batch form and the lineage formset are stricter about *contradiction*. A
+batch's status is not a free field: recording a QC result sets it to QC Pass or QC
+Fail, so the form refuses to let a person type over a quality failure. And an
+intake's weight is finite, so the formset refuses to allocate more of it than was
+received, counting every batch that has already drawn from it.
+
+Both reasons are stated in the docstrings of the classes themselves, because a
+maintainer's instinct is otherwise to "fix" them back to match the model.
 """
 from django import forms
+from django.db.models import Q, Sum
+from django.forms import modelformset_factory
 
-from apps.compliance.models import CollectorGroup, HarvestIntake, OriginZone
+from apps.batches.models import ProcessingBatch
+from apps.compliance import services
+from apps.compliance.models import (
+    CollectorGroup,
+    Facility,
+    HarvestIntake,
+    IntakeBatchLineage,
+    OriginZone,
+)
 
 # The visual grades the product specification defines.
 GRADE_CHOICES = [("A", "Grade A"), ("B", "Grade B"), ("C", "Grade C")]
@@ -154,3 +173,340 @@ class HarvestIntakeForm(forms.ModelForm):
                 ),
             )
         return cleaned
+
+
+# ---------------------------------------------------------------------------
+# Batches
+#
+# The batch form exists for a different reason from the intake form: not to stop
+# incomplete records, but to stop **contradictory** ones. A batch's status is not
+# purely a human field — recording a QC result sets it to QC Pass or QC Fail — so
+# a free status dropdown would let someone quietly type over a quality failure.
+# The rules below are the whole point of the screen; read them before changing
+# them.
+
+# The statuses the platform sets from a QC result, and which a person therefore
+# must not be able to type.
+QC_SET_STATUSES = [
+    ProcessingBatch.Status.QC_PASS,
+    ProcessingBatch.Status.QC_FAIL,
+]
+
+# ...and of those, the one that is sticky: a failed batch cannot be moved out of
+# its failure by hand at all, only by a new result. QC Pass is not sticky, because
+# a passed batch is exactly the batch that goes to sea.
+STICKY_STATUSES = [ProcessingBatch.Status.QC_FAIL]
+
+# The journey statuses, which a person does move by hand. The real world is
+# untidy: a batch can genuinely go back to processing.
+JOURNEY_STATUSES = [
+    ProcessingBatch.Status.HARVESTED,
+    ProcessingBatch.Status.PROCESSING,
+    ProcessingBatch.Status.READY_TO_SHIP,
+    ProcessingBatch.Status.IN_TRANSIT,
+    ProcessingBatch.Status.AT_UK_PORT,
+    ProcessingBatch.Status.CUSTOMS,
+    ProcessingBatch.Status.DELIVERED,
+]
+
+# A batch cannot be born mid-journey: it exists because material was processed.
+OPENING_STATUSES = [ProcessingBatch.Status.HARVESTED, ProcessingBatch.Status.PROCESSING]
+
+
+class BatchForm(forms.ModelForm):
+    """Create or correct a processing batch.
+
+    Two rules worth stating plainly, because they are the difference between a
+    working screen and a dangerous one:
+
+    - **QC Fail is sticky.** A batch sitting on a failed QC result cannot be
+      moved by hand at all. Only a new result — a retest — takes it off. To let
+      the portal move it would be to give the screen a way to make a quality
+      failure disappear.
+    - **QC Pass can move on.** It is offered as the current value (so saving
+      alone changes nothing) alongside the journey statuses, because a batch that
+      has passed QC is exactly the batch that goes to sea.
+    """
+
+    status = forms.ChoiceField(
+        choices=[],
+        label="Status",
+        help_text="Where the batch is in its journey.",
+        widget=forms.Select(attrs={"class": "form-select"}),
+    )
+    # Declared here so the required field opens on a plain instruction rather than
+    # Django's bare "---------".
+    grade = forms.ChoiceField(
+        choices=[("", "Select the grade")] + GRADE_CHOICES,
+        label="Grade",
+        help_text="The visual grade recorded for the batch, as defined by the product specification.",
+        widget=forms.Select(attrs={"class": "form-select"}),
+    )
+
+    class Meta:
+        model = ProcessingBatch
+        fields = [
+            "batch_reference",
+            "grade",
+            "weight_kg",
+            "harvest_date",
+            "origin_facility",
+            "notes",
+            "status",
+        ]
+        labels = {
+            "batch_reference": "Batch reference",
+            "weight_kg": "Weight (kg)",
+            "harvest_date": "Harvest date",
+            "origin_facility": "Origin facility",
+            "notes": "Notes",
+        }
+        help_texts = {
+            "batch_reference": "Must be unique. Suggested from the last batch recorded, and can be overwritten.",
+            "weight_kg": "The batch weight as processed, in kilograms.",
+            "harvest_date": "The date the underlying harvest was taken.",
+            "origin_facility": "The processing site. New batches choose from the facility register; an existing batch recorded against a site that is no longer registered keeps its own value, flagged so the drift is visible.",
+            "notes": "Anything the next person needs to know about this batch.",
+        }
+        widgets = {
+            "batch_reference": forms.TextInput(
+                attrs={"class": "form-control", "autocomplete": "off"}
+            ),
+            "grade": forms.Select(attrs={"class": "form-select"}),
+            "weight_kg": forms.NumberInput(
+                attrs={"class": "form-control", "step": "0.01", "min": "0.01"}
+            ),
+            "harvest_date": forms.DateInput(
+                attrs={"class": "form-control", "type": "date"}
+            ),
+            "notes": forms.Textarea(attrs={"class": "form-control", "rows": 4}),
+        }
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        # `_state.adding`, not `pk is None`: ProcessingBatch's primary key is a
+        # UUIDField with a default, so an unsaved instance already has a pk, and
+        # `bool(instance.pk)` reads as "editing" for a brand-new batch.
+        is_new = self.instance._state.adding
+        locked = not is_new and self.instance.status in STICKY_STATUSES
+
+        if locked:
+            # Shown, not editable. `disabled` means Django validates the initial
+            # value, so the current status is the only choice on offer.
+            self.fields["status"].choices = [
+                (self.instance.status, self.instance.get_status_display())
+            ]
+            self.fields["status"].disabled = True
+            self.fields["status"].help_text = (
+                "Set by the QC result and cannot be changed here. A new result — "
+                "a retest — is what moves it."
+            )
+        elif not is_new:
+            choices = list(JOURNEY_STATUSES)
+            if self.instance.status == ProcessingBatch.Status.QC_PASS:
+                # Offered so that saving alone leaves it alone, and so it can be
+                # moved on to the journey statuses.
+                choices.insert(0, ProcessingBatch.Status.QC_PASS)
+            self.fields["status"].choices = [(s.value, s.label) for s in choices]
+        else:
+            self.fields["status"].choices = [
+                (s.value, s.label) for s in OPENING_STATUSES
+            ]
+            self.fields["status"].initial = ProcessingBatch.Status.HARVESTED
+
+        # The origin facility comes from the register. An existing batch whose
+        # recorded name is not in the register keeps its own value as a choice —
+        # history is not silently rewritten — but it is labelled as the odd one
+        # out so the drift is visible.
+        current = "" if is_new else (self.instance.origin_facility or "").strip()
+        registered = [
+            f.name for f in Facility.objects.filter(is_active=True).order_by("name")
+        ]
+        choices = [(name, name) for name in registered]
+        if current and current not in registered:
+            choices.append((current, f"{current} — not in the facility register"))
+        self.fields["origin_facility"] = forms.ChoiceField(
+            choices=choices,
+            required=True,
+            label="Origin facility",
+            help_text=self.Meta.help_texts["origin_facility"],
+            widget=forms.Select(attrs={"class": "form-select"}),
+        )
+        self.fields["origin_facility"].initial = current or (
+            registered[0] if len(registered) == 1 else ""
+        )
+
+        for name in ("grade", "weight_kg", "harvest_date"):
+            self.fields[name].required = True
+
+    def clean_batch_reference(self):
+        reference = (self.cleaned_data["batch_reference"] or "").strip()
+        if not reference:
+            raise forms.ValidationError("A batch reference is required.")
+        clashes = ProcessingBatch.objects.filter(batch_reference__iexact=reference)
+        if not self.instance._state.adding:
+            clashes = clashes.exclude(pk=self.instance.pk)
+        if clashes.exists():
+            raise forms.ValidationError(
+                f"Batch {reference} already exists. References must be unique."
+            )
+        return reference
+
+    def clean_weight_kg(self):
+        weight = self.cleaned_data["weight_kg"]
+        if weight is None:
+            raise forms.ValidationError("A weight is required.")
+        if weight <= 0:
+            raise forms.ValidationError("Weight must be greater than zero.")
+        return weight
+
+    def clean_status(self):
+        status = self.cleaned_data.get("status")
+        # A locked batch's own status arrives with the form and must be allowed
+        # through — it is only a *different* QC status that is a hand-setting.
+        unchanged = not self.instance._state.adding and status == self.instance.status
+        if status in QC_SET_STATUSES and not unchanged:
+            raise forms.ValidationError(
+                f"Batch status {ProcessingBatch.Status(status).label} is set by the "
+                "QC result, not by hand."
+            )
+        if self.instance._state.adding and status not in OPENING_STATUSES:
+            raise forms.ValidationError(
+                "A new batch starts at Harvested or Processing."
+            )
+        return status
+
+
+class IntakeChoiceField(forms.ModelChoiceField):
+    """Intake dropdown that carries the facts needed to allocate from it.
+
+    An operator deciding how much of an intake can go into a batch needs the
+    weight received and what is left of it. Both are read off annotations added by
+    the view, so this costs no extra queries.
+
+    Kept short on purpose: a native ``<select>`` sizes itself to its longest
+    option, so a long label — the collector group's full name, say — widens the
+    column past the card and pushes the allocated-weight field out of sight.
+    """
+
+    def label_from_instance(self, intake):
+        parts = [intake.intake_reference]
+        if intake.weight_kg is not None:
+            parts.append(f"{intake.weight_kg} kg received")
+        elsewhere = getattr(intake, "allocated_elsewhere", None)
+        if elsewhere and intake.weight_kg is not None:
+            left = intake.weight_kg - elsewhere
+            parts.append(f"{left} kg left" if left > 0 else "fully allocated")
+        return " — ".join(parts)
+
+
+class BatchLineageForm(forms.ModelForm):
+    """One intake→batch allocation."""
+
+    harvest_intake = IntakeChoiceField(
+        queryset=HarvestIntake.objects.none(),
+        label="Harvest intake",
+        empty_label="Select the harvest intake",
+    )
+
+    class Meta:
+        model = IntakeBatchLineage
+        fields = ["harvest_intake", "weight_allocated_kg"]
+        labels = {"weight_allocated_kg": "Allocated weight (kg)"}
+        help_texts = {
+            "weight_allocated_kg": "How much of this intake went into this batch."
+        }
+        widgets = {
+            "weight_allocated_kg": forms.NumberInput(
+                attrs={"class": "form-control", "step": "0.01", "min": "0.01"}
+            ),
+        }
+
+    def __init__(self, *args, batch=None, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.batch = batch
+        queryset = HarvestIntake.objects.select_related(
+            "collector_group", "origin_zone"
+        ).order_by("intake_reference")
+        if batch is not None:
+            # What every *other* batch has taken from this intake, so the
+            # dropdown can say so and the formset can refuse the over-allocation.
+            queryset = queryset.annotate(
+                allocated_elsewhere=Sum(
+                    "batch_lineage__weight_allocated_kg",
+                    filter=~Q(batch_lineage__batch=batch),
+                )
+            )
+        self.fields["harvest_intake"].queryset = queryset
+        self.fields["weight_allocated_kg"].required = True
+
+    def clean_weight_allocated_kg(self):
+        weight = self.cleaned_data["weight_allocated_kg"]
+        if weight is None:
+            raise forms.ValidationError("An allocated weight is required.")
+        if weight <= 0:
+            raise forms.ValidationError("Allocated weight must be greater than zero.")
+        return weight
+
+
+class BaseBatchLineageFormSet(forms.BaseModelFormSet):
+    """Refuses allocations that the intake cannot support.
+
+    The model has no such rule — ``weight_allocated_kg`` is a bare decimal — which
+    is how the seeded data ended up allocating 800 kg of an 820 kg intake into two
+    different batches. The rule enforced here counts **every** batch, not just
+    this one, because the intake is a finite quantity of shea and the whole point
+    of lineage is that it adds up.
+    """
+
+    def __init__(self, *args, batch=None, **kwargs):
+        self.batch = batch
+        super().__init__(*args, **kwargs)
+
+    def clean(self):
+        super().clean()
+        if any(self.errors):
+            # Per-form errors already say what is wrong; piling cross-row errors
+            # on top of them makes the screen unreadable.
+            return
+
+        rows = []  # (form, intake, weight)
+        seen = {}
+        for form in self.forms:
+            if not getattr(form, "cleaned_data", None):
+                continue
+            if form.cleaned_data.get("DELETE"):
+                continue
+            intake = form.cleaned_data.get("harvest_intake")
+            weight = form.cleaned_data.get("weight_allocated_kg")
+            if not intake or weight is None:
+                continue
+            if intake.pk in seen:
+                form.add_error(
+                    "harvest_intake",
+                    forms.ValidationError(
+                        f"{intake.intake_reference} is already linked to this batch. "
+                        "Each intake appears once — add the two allocations together."
+                    ),
+                )
+                continue
+            seen[intake.pk] = True
+            rows.append((form, intake, weight))
+
+        for form, intake, weight in rows:
+            message = services.intake_allocation_error(
+                intake, weight, exclude_batch=self.batch
+            )
+            if message:
+                form.add_error(
+                    "weight_allocated_kg", forms.ValidationError(message)
+                )
+
+
+BatchLineageFormSet = modelformset_factory(
+    IntakeBatchLineage,
+    form=BatchLineageForm,
+    formset=BaseBatchLineageFormSet,
+    extra=1,
+    can_delete=True,
+)
diff --git a/apps/ops/tests.py b/apps/ops/tests.py
index 187b0bd..4a8b860 100644
--- a/apps/ops/tests.py
+++ b/apps/ops/tests.py
@@ -1,18 +1,28 @@
-"""The ops intake screens.
+"""The ops intake and batch screens.
 
 Two things matter here and neither is cosmetic: that only operations can create
-intake records, and that the form refuses the incomplete records the platform's
-own checks exist to catch. The model tolerates a missing date, weight or zone
-because records also arrive by API; the portal form does not.
+records, and that the forms refuse what the platform's own checks exist to catch.
+The model tolerates a missing intake date, weight or zone because records also
+arrive by API — the portal form does not. And the model has no opinion at all
+about a batch's status or about how much of an intake can be allocated, which is
+where the real rules live: a QC failure must not be typable, and an intake cannot
+give away more shea than it received.
 """
 from decimal import Decimal
 
 from django.contrib.auth import get_user_model
 from django.test import Client, TestCase
 
-from apps.compliance.models import CollectorGroup, Facility, HarvestIntake, OriginZone
+from apps.batches.models import ProcessingBatch
+from apps.compliance.models import (
+    CollectorGroup,
+    Facility,
+    HarvestIntake,
+    IntakeBatchLineage,
+    OriginZone,
+)
 
-from .views import suggested_intake_reference
+from .views import suggested_batch_reference, suggested_intake_reference
 
 User = get_user_model()
 
@@ -213,3 +223,427 @@ class SuggestedReferenceTests(IntakeTestCase):
             weight_kg=Decimal("10.00"), grade="A",
         )
         self.assertEqual(suggested_intake_reference(), "IXG-INTAKE-2026-9002")
+
+
+class BatchTestCase(TestCase):
+    """A zone, a group, a facility and three intakes — the world a batch lives in."""
+
+    def setUp(self):
+        self.ops = User.objects.create_user(
+            email="batch-ops@ixg.local", password="pass1234", role="ops",
+            full_name="Batch Officer", is_staff=True,
+        )
+        self.buyer = User.objects.create_user(
+            email="batch-buyer@ixg.local", password="pass1234", role="buyer",
+            full_name="Batch Buyer", is_staff=True,
+        )
+        self.facility = Facility.objects.create(
+            name="IXG Test Processing Facility", country="Nigeria",
+            region="Katsina State",
+        )
+        self.zone = OriginZone.objects.create(
+            facility=self.facility, zone_code="NG-KT-ZBATCH", country="Nigeria",
+            region="Katsina State", verification_status="verified",
+        )
+        self.group = CollectorGroup.objects.create(
+            name="Batch Test Collective", contact_reference="CG-BATCH-001",
+            origin_zone=self.zone,
+        )
+        self.intakes = [
+            HarvestIntake.objects.create(
+                intake_reference=f"IXG-INTAKE-2026-91{n:02d}", collector_group=self.group,
+                origin_zone=self.zone, intake_date=f"2026-09-0{n}",
+                weight_kg=Decimal(weight), grade="A", status="verified",
+            )
+            for n, weight in ((1, "820.00"), (2, "640.00"), (3, "600.00"))
+        ]
+        self.client_ops = Client(SERVER_NAME="ops.ixgltd.com")
+        self.client_ops.force_login(self.ops)
+        self.client_buyer = Client(SERVER_NAME="ops.ixgltd.com")
+        self.client_buyer.force_login(self.buyer)
+        self.batch = ProcessingBatch.objects.create(
+            batch_reference="IXG-BATCH-2026-901", harvest_date="2026-09-04",
+            weight_kg=Decimal("1500.00"), grade="A", status="harvested",
+            origin_facility=self.facility.name, created_by=self.ops,
+        )
+
+    def payload(self, **overrides):
+        data = {
+            "batch_reference": "IXG-BATCH-2026-902",
+            "grade": "A",
+            "weight_kg": "1200.00",
+            "harvest_date": "2026-09-10",
+            "origin_facility": self.facility.name,
+            "notes": "Synthetic batch for tests.",
+            "status": "harvested",
+        }
+        data.update(overrides)
+        return data
+
+    def errors_for(self, response, field=None):
+        form = response.context["form"]
+        if field:
+            return " ".join(form.errors.get(field, []))
+        return " ".join(str(v) for v in form.errors.values())
+
+    def lineage_payload(self, rows, initial=0):
+        """A formset POST body. Each row: {intake, weight, id?, delete?}."""
+        data = {
+            "lineage-TOTAL_FORMS": str(len(rows) + 1),
+            "lineage-INITIAL_FORMS": str(initial),
+            "lineage-MIN_NUM_FORMS": "0",
+            "lineage-MAX_NUM_FORMS": "1000",
+        }
+        for i, row in enumerate(rows):
+            data[f"lineage-{i}-harvest_intake"] = str(row.get("intake") or "")
+            data[f"lineage-{i}-weight_allocated_kg"] = str(row.get("weight") or "")
+            if row.get("id"):
+                data[f"lineage-{i}-id"] = str(row["id"])
+            if row.get("delete"):
+                data[f"lineage-{i}-DELETE"] = "on"
+        data[f"lineage-{len(rows)}-harvest_intake"] = ""
+        data[f"lineage-{len(rows)}-weight_allocated_kg"] = ""
+        return data
+
+    def lineage_errors(self, response):
+        formset = response.context["formset"]
+        return " ".join(
+            str(e) for form in formset.forms for e in form.errors.values()
+        )
+
+
+class BatchAccessTests(BatchTestCase):
+    def test_ops_can_open_every_batch_screen(self):
+        for path in (
+            "/ops/batches/", "/ops/batches/new/",
+            f"/ops/batches/{self.batch.pk}/edit/",
+            f"/ops/batches/{self.batch.pk}/lineage/",
+        ):
+            with self.subTest(path=path):
+                self.assertEqual(self.client_ops.get(path).status_code, 200)
+
+    def test_a_buyer_cannot_open_them(self):
+        for path in (
+            "/ops/batches/new/", f"/ops/batches/{self.batch.pk}/edit/",
+            f"/ops/batches/{self.batch.pk}/lineage/",
+        ):
+            with self.subTest(path=path):
+                self.assertEqual(self.client_buyer.get(path).status_code, 403)
+
+    def test_a_buyer_cannot_post_a_new_batch(self):
+        r = self.client_buyer.post("/ops/batches/new/", self.payload())
+        self.assertEqual(r.status_code, 403)
+        self.assertFalse(
+            ProcessingBatch.objects.filter(batch_reference="IXG-BATCH-2026-902").exists()
+        )
+
+    def test_a_buyer_cannot_post_an_edit(self):
+        r = self.client_buyer.post(
+            f"/ops/batches/{self.batch.pk}/edit/",
+            self.payload(batch_reference="IXG-BATCH-2026-901", status="delivered"),
+        )
+        self.assertEqual(r.status_code, 403)
+        self.batch.refresh_from_db()
+        self.assertEqual(self.batch.status, "harvested")
+
+    def test_the_list_offers_a_new_batch_control(self):
+        r = self.client_ops.get("/ops/batches/")
+        self.assertContains(r, "New batch")
+        self.assertIn("/ops/batches/new/", r.content.decode())
+
+
+class BatchCreateTests(BatchTestCase):
+    def test_a_complete_batch_is_created_and_sent_to_its_lineage(self):
+        r = self.client_ops.post("/ops/batches/new/", self.payload())
+        self.assertEqual(r.status_code, 302)
+        batch = ProcessingBatch.objects.get(batch_reference="IXG-BATCH-2026-902")
+        self.assertEqual(batch.weight_kg, Decimal("1200.00"))
+        self.assertEqual(batch.origin_facility, self.facility.name)
+        # Created by the signed-in user, and pointed straight at the lineage,
+        # because a batch with no intake behind it is not traceable.
+        self.assertEqual(batch.created_by, self.ops)
+        self.assertEqual(r["Location"], f"/ops/batches/{batch.pk}/lineage/")
+
+    def test_the_form_suggests_the_next_reference(self):
+        r = self.client_ops.get("/ops/batches/new/")
+        self.assertContains(r, f'value="{suggested_batch_reference()}"')
+
+    def test_a_duplicate_reference_is_refused(self):
+        r = self.client_ops.post(
+            "/ops/batches/new/", self.payload(batch_reference="IXG-BATCH-2026-901")
+        )
+        self.assertEqual(r.status_code, 200)
+        self.assertIn("already exists", self.errors_for(r, "batch_reference"))
+
+    def test_a_duplicate_differing_only_in_case_is_refused(self):
+        r = self.client_ops.post(
+            "/ops/batches/new/", self.payload(batch_reference="ixg-batch-2026-901")
+        )
+        self.assertIn("already exists", self.errors_for(r, "batch_reference"))
+
+    def test_a_zero_or_negative_weight_is_refused(self):
+        for weight in ("0", "-5"):
+            with self.subTest(weight=weight):
+                r = self.client_ops.post("/ops/batches/new/", self.payload(weight_kg=weight))
+                self.assertIn("greater than zero", self.errors_for(r, "weight_kg"))
+
+    def test_the_fields_a_batch_needs_are_required(self):
+        for field in ("batch_reference", "grade", "weight_kg", "harvest_date",
+                      "origin_facility", "status"):
+            with self.subTest(field=field):
+                r = self.client_ops.post("/ops/batches/new/", self.payload(**{field: ""}))
+                self.assertEqual(r.status_code, 200)
+                self.assertTrue(
+                    self.errors_for(r, field), f"{field} was accepted blank"
+                )
+
+    def test_a_new_batch_cannot_be_born_mid_journey(self):
+        for status in ("in_transit", "delivered", "ready_to_ship"):
+            with self.subTest(status=status):
+                r = self.client_ops.post("/ops/batches/new/", self.payload(status=status))
+                self.assertTrue(self.errors_for(r, "status"))
+
+    def test_a_qc_status_cannot_be_typed_by_hand(self):
+        for status in ("qc_pass", "qc_fail"):
+            with self.subTest(status=status):
+                r = self.client_ops.post("/ops/batches/new/", self.payload(status=status))
+                self.assertTrue(self.errors_for(r, "status"))
+                self.assertFalse(
+                    ProcessingBatch.objects.filter(batch_reference="IXG-BATCH-2026-902").exists()
+                )
+
+    def test_the_origin_facility_comes_from_the_register(self):
+        r = self.client_ops.post(
+            "/ops/batches/new/", self.payload(origin_facility="A site nobody registered")
+        )
+        self.assertTrue(self.errors_for(r, "origin_facility"))
+
+
+class BatchStatusRuleTests(BatchTestCase):
+    """The rule that makes the screen worth having."""
+
+    def test_a_journey_status_is_moved_by_hand(self):
+        r = self.client_ops.post(
+            f"/ops/batches/{self.batch.pk}/edit/",
+            self.payload(batch_reference="IXG-BATCH-2026-901", status="ready_to_ship"),
+        )
+        self.assertEqual(r.status_code, 302)
+        self.batch.refresh_from_db()
+        self.assertEqual(self.batch.status, "ready_to_ship")
+
+    def test_a_passed_batch_can_go_on_to_ship(self):
+        self.batch.status = "qc_pass"
+        self.batch.save()
+        r = self.client_ops.post(
+            f"/ops/batches/{self.batch.pk}/edit/",
+            self.payload(batch_reference="IXG-BATCH-2026-901", status="ready_to_ship"),
+        )
+        self.assertEqual(r.status_code, 302)
+        self.batch.refresh_from_db()
+        self.assertEqual(self.batch.status, "ready_to_ship")
+
+    def test_a_failed_batch_keeps_its_failed_status(self):
+        """The screen must not be a way to make a quality failure disappear."""
+        self.batch.status = "qc_fail"
+        self.batch.save()
+        r = self.client_ops.get(f"/ops/batches/{self.batch.pk}/edit/")
+        html = r.content.decode()
+        self.assertIn("Set by the QC result", html)
+        # Only the current status is on offer, so it cannot be typed over.
+        self.assertNotIn('value="ready_to_ship"', html)
+        self.assertNotIn('value="delivered"', html)
+
+        r = self.client_ops.post(
+            f"/ops/batches/{self.batch.pk}/edit/",
+            self.payload(batch_reference="IXG-BATCH-2026-901", status="ready_to_ship"),
+        )
+        self.batch.refresh_from_db()
+        self.assertEqual(self.batch.status, "qc_fail")
+        self.assertEqual(r.status_code, 302)
+
+    def test_a_qc_status_cannot_be_hand_set_on_a_journey_batch(self):
+        r = self.client_ops.post(
+            f"/ops/batches/{self.batch.pk}/edit/",
+            self.payload(batch_reference="IXG-BATCH-2026-901", status="qc_fail"),
+        )
+        self.assertTrue(self.errors_for(r, "status"))
+        self.batch.refresh_from_db()
+        self.assertEqual(self.batch.status, "harvested")
+
+    def test_editing_updates_in_place(self):
+        r = self.client_ops.post(
+            f"/ops/batches/{self.batch.pk}/edit/",
+            self.payload(batch_reference="IXG-BATCH-2026-901", weight_kg="1400.00"),
+        )
+        self.assertEqual(r.status_code, 302)
+        self.batch.refresh_from_db()
+        self.assertEqual(self.batch.weight_kg, Decimal("1400.00"))
+        self.assertEqual(ProcessingBatch.objects.count(), 1)
+
+    def test_an_existing_facility_that_is_not_registered_is_kept_and_flagged(self):
+        """History is not silently rewritten, but the drift is made visible."""
+        self.batch.origin_facility = "ASI - Katsina State, Nigeria"
+        self.batch.save()
+        r = self.client_ops.get(f"/ops/batches/{self.batch.pk}/edit/")
+        self.assertContains(r, "not in the facility register")
+
+    def test_saving_under_its_own_reference_is_not_a_duplicate(self):
+        r = self.client_ops.post(
+            f"/ops/batches/{self.batch.pk}/edit/",
+            self.payload(batch_reference="IXG-BATCH-2026-901"),
+        )
+        self.assertEqual(r.status_code, 302)
+
+
+class BatchLineageTests(BatchTestCase):
+    def test_the_screen_shows_what_is_allocated_and_the_batch_weight(self):
+        IntakeBatchLineage.objects.create(
+            batch=self.batch, harvest_intake=self.intakes[0],
+            weight_allocated_kg=Decimal("500.00"),
+        )
+        r = self.client_ops.get(f"/ops/batches/{self.batch.pk}/lineage/")
+        self.assertContains(r, "500.00")
+        self.assertContains(r, "1500.00")
+
+    def test_an_allocation_is_created_against_the_batch(self):
+        r = self.client_ops.post(
+            f"/ops/batches/{self.batch.pk}/lineage/",
+            self.lineage_payload([{"intake": self.intakes[0].pk, "weight": "800.00"}]),
+        )
+        self.assertEqual(r.status_code, 302)
+        row = IntakeBatchLineage.objects.get(batch=self.batch)
+        self.assertEqual(row.harvest_intake, self.intakes[0])
+        self.assertEqual(row.weight_allocated_kg, Decimal("800.00"))
+
+    def test_the_batch_detail_page_shows_the_lineage(self):
+        IntakeBatchLineage.objects.create(
+            batch=self.batch, harvest_intake=self.intakes[2],
+            weight_allocated_kg=Decimal("600.00"),
+        )
+        r = self.client_ops.get(f"/ops/batches/{self.batch.pk}/")
+        self.assertContains(r, "Intake Lineage")
+        self.assertContains(r, self.intakes[2].intake_reference)
+        self.assertContains(r, "600.00")
+
+    def test_a_batch_with_no_lineage_is_told_so(self):
+        r = self.client_ops.get(f"/ops/batches/{self.batch.pk}/")
+        self.assertContains(r, "No harvest intakes are linked to this batch")
+
+    def test_more_than_the_intake_received_is_refused(self):
+        r = self.client_ops.post(
+            f"/ops/batches/{self.batch.pk}/lineage/",
+            self.lineage_payload([{"intake": self.intakes[0].pk, "weight": "900.00"}]),
+        )
+        self.assertEqual(r.status_code, 200)
+        self.assertIn("cannot be allocated", self.lineage_errors(r))
+        self.assertFalse(IntakeBatchLineage.objects.exists())
+
+    def test_an_intake_already_taken_by_another_batch_is_refused(self):
+        """The rule that counts every batch — and the defect in the old seed data."""
+        other = ProcessingBatch.objects.create(
+            batch_reference="IXG-BATCH-2026-903", weight_kg=Decimal("800.00"),
+            grade="A", status="processing", origin_facility=self.facility.name,
+        )
+        IntakeBatchLineage.objects.create(
+            batch=other, harvest_intake=self.intakes[0],
+            weight_allocated_kg=Decimal("800.00"),
+        )
+        r = self.client_ops.post(
+            f"/ops/batches/{self.batch.pk}/lineage/",
+            self.lineage_payload([{"intake": self.intakes[0].pk, "weight": "800.00"}]),
+        )
+        self.assertEqual(r.status_code, 200)
+        message = self.lineage_errors(r)
+        self.assertIn("already allocated", message)
+        self.assertIn("IXG-BATCH-2026-903", message)
+        self.assertEqual(IntakeBatchLineage.objects.filter(batch=self.batch).count(), 0)
+
+    def test_what_is_left_of_an_intake_can_still_be_allocated(self):
+        other = ProcessingBatch.objects.create(
+            batch_reference="IXG-BATCH-2026-904", weight_kg=Decimal("500.00"),
+            grade="A", status="processing", origin_facility=self.facility.name,
+        )
+        IntakeBatchLineage.objects.create(
+            batch=other, harvest_intake=self.intakes[0],
+            weight_allocated_kg=Decimal("500.00"),
+        )
+        r = self.client_ops.post(
+            f"/ops/batches/{self.batch.pk}/lineage/",
+            self.lineage_payload([{"intake": self.intakes[0].pk, "weight": "300.00"}]),
+        )
+        self.assertEqual(r.status_code, 302)
+        self.assertEqual(
+            IntakeBatchLineage.objects.get(batch=self.batch).weight_allocated_kg,
+            Decimal("300.00"),
+        )
+
+    def test_the_same_intake_twice_in_one_batch_is_refused(self):
+        r = self.client_ops.post(
+            f"/ops/batches/{self.batch.pk}/lineage/",
+            self.lineage_payload([
+                {"intake": self.intakes[0].pk, "weight": "200.00"},
+                {"intake": self.intakes[0].pk, "weight": "300.00"},
+            ]),
+        )
+        self.assertEqual(r.status_code, 200)
+        self.assertIn("already linked to this batch", self.lineage_errors(r))
+
+    def test_a_zero_allocation_is_refused(self):
+        r = self.client_ops.post(
+            f"/ops/batches/{self.batch.pk}/lineage/",
+            self.lineage_payload([{"intake": self.intakes[0].pk, "weight": "0"}]),
+        )
+        self.assertTrue(self.lineage_errors(r))
+
+    def test_an_allocation_can_be_corrected_and_removed(self):
+        row = IntakeBatchLineage.objects.create(
+            batch=self.batch, harvest_intake=self.intakes[0],
+            weight_allocated_kg=Decimal("500.00"),
+        )
+        r = self.client_ops.post(
+            f"/ops/batches/{self.batch.pk}/lineage/",
+            self.lineage_payload(
+                [{"id": row.pk, "intake": self.intakes[0].pk, "weight": "520.00"}],
+                initial=1,
+            ),
+        )
+        self.assertEqual(r.status_code, 302)
+        row.refresh_from_db()
+        self.assertEqual(row.weight_allocated_kg, Decimal("520.00"))
+
+        r = self.client_ops.post(
+            f"/ops/batches/{self.batch.pk}/lineage/",
+            self.lineage_payload(
+                [{"id": row.pk, "intake": self.intakes[0].pk, "weight": "520.00",
+                  "delete": True}],
+                initial=1,
+            ),
+        )
+        self.assertEqual(r.status_code, 302)
+        self.assertFalse(IntakeBatchLineage.objects.filter(batch=self.batch).exists())
+        # The intake itself is untouched: removing a link is not deleting material.
+        self.assertTrue(HarvestIntake.objects.filter(pk=self.intakes[0].pk).exists())
+
+    def test_a_buyer_cannot_post_an_allocation(self):
+        r = self.client_buyer.post(
+            f"/ops/batches/{self.batch.pk}/lineage/",
+            self.lineage_payload([{"intake": self.intakes[0].pk, "weight": "500.00"}]),
+        )
+        self.assertEqual(r.status_code, 403)
+        self.assertFalse(IntakeBatchLineage.objects.exists())
+
+
+class SuggestedBatchReferenceTests(BatchTestCase):
+    def test_the_suggestion_continues_the_year_sequence(self):
+        self.assertEqual(suggested_batch_reference(), "IXG-BATCH-2026-902")
+
+    def test_the_suggestion_ignores_other_years_and_odd_references(self):
+        ProcessingBatch.objects.create(
+            batch_reference="IXG-BATCH-2025-9999", weight_kg=Decimal("10.00"),
+            grade="A", status="harvested",
+        )
+        ProcessingBatch.objects.create(
+            batch_reference="IXG-TEST-001", weight_kg=Decimal("10.00"),
+            grade="A", status="harvested",
+        )
+        self.assertEqual(suggested_batch_reference(), "IXG-BATCH-2026-902")
diff --git a/apps/ops/urls.py b/apps/ops/urls.py
index 0f2a035..ef8474e 100644
--- a/apps/ops/urls.py
+++ b/apps/ops/urls.py
@@ -6,7 +6,10 @@ app_name = 'ops'
 urlpatterns = [
     path('', views.DashboardView.as_view(), name='dashboard'),
     path('batches/', views.BatchListView.as_view(), name='batch_list'),
+    path('batches/new/', views.BatchCreateView.as_view(), name='batch_create'),
     path('batches/<uuid:pk>/', views.BatchDetailView.as_view(), name='batch_detail'),
+    path('batches/<uuid:pk>/edit/', views.BatchUpdateView.as_view(), name='batch_update'),
+    path('batches/<uuid:pk>/lineage/', views.BatchLineageView.as_view(), name='batch_lineage'),
     path('intakes/', views.IntakeListView.as_view(), name='intake_list'),
     path('intakes/new/', views.IntakeCreateView.as_view(), name='intake_create'),
     path('intakes/<uuid:pk>/edit/', views.IntakeUpdateView.as_view(), name='intake_update'),
diff --git a/apps/ops/views.py b/apps/ops/views.py
index 1d2b394..ab05545 100644
--- a/apps/ops/views.py
+++ b/apps/ops/views.py
@@ -8,7 +8,7 @@ from django.urls import reverse
 from django.utils import timezone
 from django.views.generic import TemplateView, ListView, DetailView, CreateView, UpdateView
 from django.views import View
-from django.shortcuts import get_object_or_404
+from django.shortcuts import get_object_or_404, render
 
 from apps.batches.models import ProcessingBatch
 from apps.qc.models import QCResult
@@ -18,10 +18,15 @@ from apps.documents.downloads import file_response
 from apps.documents.models import Document, CertificateOfAnalysis
 from apps.esg.models import ESGFieldReport
 from apps.notifications.models import Notification
-from apps.compliance.models import CollectorGroup, DueDiligenceCase, HarvestIntake
+from apps.compliance.models import (
+    CollectorGroup,
+    DueDiligenceCase,
+    HarvestIntake,
+    IntakeBatchLineage,
+)
 from apps.guides.views import GuideDetailView as BaseGuideDetailView
 from apps.guides.views import GuideIndexView as BaseGuideIndexView
-from .forms import HarvestIntakeForm
+from .forms import BatchForm, BatchLineageFormSet, HarvestIntakeForm
 
 
 def suggested_intake_reference(today=None):
@@ -42,6 +47,24 @@ def suggested_intake_reference(today=None):
     return f"{prefix}{highest + 1:04d}"
 
 
+def suggested_batch_reference(today=None):
+    """Next batch reference, continuing this year's sequence. Same reasoning.
+
+    Three digits, because that is the shape the existing references have
+    (``IXG-BATCH-2026-014``); it widens by itself past 999.
+    """
+    year = (today or timezone.localdate()).year
+    prefix = f"IXG-BATCH-{year}-"
+    highest = 0
+    for ref in ProcessingBatch.objects.filter(
+        batch_reference__startswith=prefix
+    ).values_list("batch_reference", flat=True):
+        tail = ref[len(prefix):]
+        if tail.isdigit():
+            highest = max(highest, int(tail))
+    return f"{prefix}{highest + 1:03d}"
+
+
 class OpsRequiredMixin(UserPassesTestMixin):
     def test_func(self):
         user = self.request.user
@@ -114,9 +137,150 @@ class BatchDetailView(OpsRequiredMixin, DetailView):
         ctx['capa_records'] = CAPARecord.objects.filter(batch=batch)
         ctx['documents'] = Document.objects.filter(batch=batch)
         ctx['esg_reports'] = ESGFieldReport.objects.filter(batch=batch)
+        lineage = batch.intake_lineage.select_related(
+            'harvest_intake',
+            'harvest_intake__origin_zone',
+            'harvest_intake__collector_group',
+        ).order_by('harvest_intake__intake_reference')
+        ctx['lineage'] = lineage
+        ctx['lineage_allocated'] = sum((row.weight_allocated_kg or 0) for row in lineage)
+        if batch.weight_kg is not None:
+            ctx['lineage_difference'] = batch.weight_kg - ctx['lineage_allocated']
         return ctx
 
 
+class BatchCreateView(OpsRequiredMixin, CreateView):
+    model = ProcessingBatch
+    form_class = BatchForm
+    template_name = 'ops/batch_form.html'
+
+    def get_initial(self):
+        initial = super().get_initial()
+        initial['batch_reference'] = suggested_batch_reference()
+        initial['harvest_date'] = timezone.localdate()
+        initial['status'] = ProcessingBatch.Status.HARVESTED
+        return initial
+
+    def get_context_data(self, **kwargs):
+        ctx = super().get_context_data(**kwargs)
+        ctx['page_title'] = 'New processing batch'
+        ctx['submit_label'] = 'Create batch'
+        # The template cannot ask the instance whether it is new — ProcessingBatch's
+        # primary key is a UUID with a default, so an unsaved batch already has one,
+        # and `_state` is unreachable from a template. The view knows, so it says.
+        ctx['creating'] = True
+        return ctx
+
+    def form_valid(self, form):
+        form.instance.created_by = self.request.user
+        response = super().form_valid(form)
+        messages.success(
+            self.request,
+            f"Batch {self.object.batch_reference} created. "
+            "Allocate its harvest intakes next — a batch with no lineage fails "
+            "the due-diligence lineage check.",
+        )
+        return response
+
+    def get_success_url(self):
+        # Straight to the lineage, because a batch with no intake behind it is
+        # not traceable and the check will say so.
+        return reverse('ops:batch_lineage', kwargs={'pk': self.object.pk})
+
+
+class BatchUpdateView(OpsRequiredMixin, UpdateView):
+    model = ProcessingBatch
+    form_class = BatchForm
+    template_name = 'ops/batch_form.html'
+
+    def get_context_data(self, **kwargs):
+        ctx = super().get_context_data(**kwargs)
+        ctx['page_title'] = f"Edit batch {self.object.batch_reference}"
+        ctx['submit_label'] = 'Save changes'
+        ctx['creating'] = False
+        return ctx
+
+    def form_valid(self, form):
+        response = super().form_valid(form)
+        messages.success(
+            self.request, f"Batch {self.object.batch_reference} updated."
+        )
+        return response
+
+    def get_success_url(self):
+        return reverse('ops:batch_detail', kwargs={'pk': self.object.pk})
+
+
+class BatchLineageView(OpsRequiredMixin, View):
+    """Allocate harvest intakes to a batch, and correct what is already there.
+
+    This is the screen that closes the lineage gap in walkthrough steps 3 and 4:
+    the link between an intake and a batch, with the weight that went into it, is
+    created here rather than in the administration area. The formset refuses an
+    allocation the intake cannot support — see
+    ``apps/ops/forms.py::BaseBatchLineageFormSet``.
+    """
+
+    template_name = 'ops/batch_lineage.html'
+
+    def get_batch(self):
+        return get_object_or_404(ProcessingBatch, pk=self.kwargs['pk'])
+
+    def get_formset(self, batch, data=None):
+        queryset = IntakeBatchLineage.objects.filter(batch=batch).select_related(
+            'harvest_intake'
+        ).order_by('harvest_intake__intake_reference')
+        return BatchLineageFormSet(
+            data,
+            queryset=queryset,
+            prefix='lineage',
+            batch=batch,
+            form_kwargs={'batch': batch},
+        )
+
+    def get_context_data(self, batch, formset):
+        rows = list(batch.intake_lineage.select_related(
+            'harvest_intake', 'harvest_intake__origin_zone',
+            'harvest_intake__collector_group',
+        ).order_by('harvest_intake__intake_reference'))
+        allocated = sum((row.weight_allocated_kg or 0) for row in rows)
+        return {
+            'batch': batch,
+            'formset': formset,
+            'lineage_rows': rows,
+            'total_allocated': allocated,
+            'difference': (batch.weight_kg - allocated)
+            if batch.weight_kg is not None else None,
+        }
+
+    def get(self, request, *args, **kwargs):
+        batch = self.get_batch()
+        formset = self.get_formset(batch)
+        return render(
+            request, self.template_name, self.get_context_data(batch, formset)
+        )
+
+    def post(self, request, *args, **kwargs):
+        batch = self.get_batch()
+        formset = self.get_formset(batch, data=request.POST)
+        if formset.is_valid():
+            # A plain ModelFormSet does not know the batch, so the new rows get it
+            # here; the deleted ones are removed explicitly.
+            for obj in formset.save(commit=False):
+                obj.batch = batch
+                obj.save()
+            for obj in formset.deleted_objects:
+                obj.delete()
+            messages.success(
+                request,
+                f"Intake lineage for {batch.batch_reference} updated.",
+            )
+            return redirect('ops:batch_detail', pk=batch.pk)
+        return render(
+            request, self.template_name, self.get_context_data(batch, formset)
+        )
+
+
 class QCListView(OpsRequiredMixin, ListView):
     model = QCResult
     template_name = 'ops/qc_list.html'
diff --git a/templates/ops/batch_form.html b/templates/ops/batch_form.html
new file mode 100644
index 0000000..3c8ec1b
--- /dev/null
+++ b/templates/ops/batch_form.html
@@ -0,0 +1,92 @@
+{% extends 'ops/base_ops.html' %}
+{% block title %}{{ page_title }} - IXG Operations{% endblock %}
+{% block content %}
+
+<div class="page-header d-flex justify-content-between align-items-center">
+  <h2>{{ page_title }}</h2>
+  <a href="{% url 'ops:batch_list' %}" class="btn btn-sm btn-outline-secondary"><i class="bi bi-arrow-left"></i> Back</a>
+</div>
+
+<div class="row g-3">
+  <div class="col-md-7">
+    <div class="stat-card">
+      <form method="post" novalidate id="batch-form">
+        {% csrf_token %}
+
+        {% if form.non_field_errors %}
+          {% for e in form.non_field_errors %}<div class="alert alert-danger py-2">{{ e }}</div>{% endfor %}
+        {% endif %}
+
+        {% for field in form %}
+        <div class="mb-3">
+          <label class="form-label" for="{{ field.id_for_label }}">
+            {{ field.label }}{% if field.field.required and not field.field.disabled %} <span class="text-danger">*</span>{% endif %}
+          </label>
+          {{ field }}
+          {% if field.help_text %}<div class="form-text">{{ field.help_text }}</div>{% endif %}
+          {% for e in field.errors %}<div class="text-danger small">{{ e }}</div>{% endfor %}
+        </div>
+        {% endfor %}
+
+        <button type="submit" class="btn btn-primary">{{ submit_label }}</button>
+        <a href="{% url 'ops:batch_list' %}" class="btn btn-outline-secondary">Cancel</a>
+      </form>
+    </div>
+  </div>
+
+  <div class="col-md-5">
+    <div class="stat-card mb-3">
+      <h6 class="mb-3"><i class="bi bi-diagram-3"></i> What a batch is</h6>
+      <p class="small mb-2">
+        The batch is the spine of the story: QC results, the certificate of
+        analysis, the due-diligence case, shipments and the buyer's view all hang
+        off it. If the reference is wrong, everything downstream is wrong with it.
+      </p>
+      <p class="small mb-0">
+        Creating a batch is only half of it. It also needs its <strong>harvest
+        intakes</strong> allocated to it, or it cannot be traced back to a
+        collector group and an origin zone.
+      </p>
+    </div>
+
+    <div class="stat-card mb-3">
+      <h6 class="mb-3"><i class="bi bi-shield-check"></i> Who sets the status</h6>
+      <ul class="small mb-2 ps-3">
+        <li><strong>Harvested, Processing, Ready to Ship, In Transit, At UK Port,
+          Customs, Delivered</strong> &mdash; moved here, by you.</li>
+        <li><strong>QC Fail</strong> &mdash; never offered here; set by the platform
+          when a QC result is recorded (step 5), because a quality failure must not
+          be something a person can type over.</li>
+        <li><strong>QC Pass</strong> &mdash; appears only as the current value when
+          editing a batch that has already passed, so saving alone leaves it alone
+          and it can be moved on to the journey statuses.</li>
+      </ul>
+      <p class="small mb-0">
+        A batch sitting on a failed result cannot be moved at all until a new
+        result &mdash; a retest &mdash; is recorded against it.
+      </p>
+    </div>
+
+    {% if not creating %}
+    <div class="stat-card">
+      <h6 class="mb-3"><i class="bi bi-basket"></i> Harvest intakes</h6>
+      <p class="small mb-2">
+        Which intakes went into this batch, and how much of each. Allocated from
+        the intakes screen for the batch.
+      </p>
+      <a href="{% url 'ops:batch_lineage' form.instance.pk %}" class="btn btn-sm btn-outline-primary">Edit intake lineage</a>
+    </div>
+    {% else %}
+    <div class="stat-card">
+      <h6 class="mb-3"><i class="bi bi-basket"></i> Harvest intakes</h6>
+      <p class="small mb-2">
+        Once the batch exists you will be taken straight to its harvest intakes,
+        so you can allocate what went into it. A batch with no intake linked to it
+        cannot be traced back to a collector group and an origin zone.
+      </p>
+    </div>
+    {% endif %}
+  </div>
+</div>
+
+{% endblock %}
diff --git a/templates/ops/batch_lineage.html b/templates/ops/batch_lineage.html
new file mode 100644
index 0000000..f073d90
--- /dev/null
+++ b/templates/ops/batch_lineage.html
@@ -0,0 +1,114 @@
+{% extends 'ops/base_ops.html' %}
+{% block title %}Intake lineage - {{ batch.batch_reference }} - IXG Operations{% endblock %}
+{% block content %}
+
+{% if messages %}
+  {% for m in messages %}<div class="alert alert-{{ m.tags }} py-2">{{ m }}</div>{% endfor %}
+{% endif %}
+
+<div class="page-header d-flex justify-content-between align-items-center">
+  <h2>Intake lineage: {{ batch.batch_reference }}</h2>
+  <a href="{% url 'ops:batch_detail' batch.pk %}" class="btn btn-sm btn-outline-secondary"><i class="bi bi-arrow-left"></i> Back to the batch</a>
+</div>
+
+<div class="row g-3">
+  <div class="col-md-8">
+    <div class="stat-card">
+      <form method="post" novalidate>
+        {% csrf_token %}
+        {{ formset.management_form }}
+
+        {% if formset.non_form_errors %}
+          {% for e in formset.non_form_errors %}<div class="alert alert-danger py-2">{{ e }}</div>{% endfor %}
+        {% endif %}
+
+        <!-- table-layout: fixed, or the intake dropdown's intrinsic width (a native
+             select sizes to its longest option) pushes the weight field and the
+             Remove box off the card. -->
+        <table class="table table-sm align-middle mb-3" style="table-layout: fixed;">
+          <thead>
+            <tr>
+              <th>Harvest intake</th>
+              <th style="width: 10rem;">Allocated (kg)</th>
+              <th style="width: 6rem;">Remove</th>
+            </tr>
+          </thead>
+          <tbody>
+            {% for f in formset.forms %}
+            <tr>
+              {% for hidden in f.hidden_fields %}{{ hidden }}{% endfor %}
+              <td>
+                {{ f.harvest_intake }}
+                {% for e in f.harvest_intake.errors %}<div class="text-danger small">{{ e }}</div>{% endfor %}
+              </td>
+              <td>
+                {{ f.weight_allocated_kg }}
+                {% for e in f.weight_allocated_kg.errors %}<div class="text-danger small">{{ e }}</div>{% endfor %}
+              </td>
+              <td>
+                {% if f.instance.pk %}{{ f.DELETE }}{% endif %}
+              </td>
+            </tr>
+            {% endfor %}
+          </tbody>
+        </table>
+
+        <button type="submit" class="btn btn-primary">Save lineage</button>
+        <a href="{% url 'ops:batch_detail' batch.pk %}" class="btn btn-outline-secondary">Cancel</a>
+      </form>
+    </div>
+  </div>
+
+  <div class="col-md-4">
+    <div class="stat-card mb-3">
+      <h6 class="mb-3"><i class="bi bi-calculator"></i> Weights</h6>
+      <table class="table table-sm mb-0">
+        <tr><td class="text-secondary">Allocated to this batch</td><td>{{ total_allocated }} kg</td></tr>
+        <tr><td class="text-secondary">Batch weight</td><td>{{ batch.weight_kg|default:'-' }} kg</td></tr>
+        {% if difference is not None %}
+        <tr>
+          <td class="text-secondary">Difference</td>
+          <td>{{ difference }} kg</td>
+        </tr>
+        {% endif %}
+      </table>
+      <p class="small text-secondary mt-2 mb-0">
+        The platform shows these two figures side by side; it does not reconcile
+        them. A difference where the batch is heavier means part of the batch is
+        not traced to an intake; allocations above the batch weight are the usual
+        processing loss.
+      </p>
+    </div>
+
+    <div class="stat-card mb-3">
+      <h6 class="mb-3"><i class="bi bi-shield-check"></i> What this feeds</h6>
+      <ul class="small mb-2 ps-3">
+        <li><strong>Lineage</strong> &mdash; the batch needs at least one intake</li>
+        <li><strong>Geolocation</strong> &mdash; every linked intake must point at a
+          <em>verified</em> origin zone</li>
+        <li><strong>Intake data</strong> &mdash; every linked intake needs a date and
+          a weight</li>
+      </ul>
+      <p class="small mb-0">
+        This table is what the due-diligence case for {{ batch.batch_reference }}
+        reads, and what a buyer or an auditor traces the batch back through.
+      </p>
+    </div>
+
+    <div class="stat-card">
+      <h6 class="mb-3"><i class="bi bi-exclamation-triangle"></i> One intake, limited weight</h6>
+      <p class="small mb-2">
+        An intake received a finite weight of shea, and what is allocated out of it
+        is counted across <strong>every</strong> batch. An allocation that would
+        take more than the intake received is refused, whether the excess is over
+        the intake's own weight or over what other batches have already taken.
+      </p>
+      <p class="small mb-0">
+        Removing a row here deletes the link only &mdash; the intake record itself
+        is untouched.
+      </p>
+    </div>
+  </div>
+</div>
+
+{% endblock %}
diff --git a/templates/ops/batch_list.html b/templates/ops/batch_list.html
index f6a7d51..d5cfc0c 100644
--- a/templates/ops/batch_list.html
+++ b/templates/ops/batch_list.html
@@ -4,7 +4,10 @@
 
 <div class="page-header d-flex justify-content-between align-items-center">
   <h2>Processing Batches</h2>
-  <span class="text-secondary">{{ page_obj.paginator.count }} total</span>
+  <div>
+    <span class="text-secondary me-2">{{ page_obj.paginator.count }} total</span>
+    <a href="{% url 'ops:batch_create' %}" class="btn btn-sm btn-primary"><i class="bi bi-plus-lg"></i> New batch</a>
+  </div>
 </div>
 
 <div class="filter-bar">