fix(selling): sum repeated components in Available Stock for Packing Items (#59506)

* fix(selling): sum repeated components in Available Stock for Packing Items

A Product Bundle can list the same component on more than one line. The
report divided the projected qty by each line's qty on its own, so a bundle
needing 2 + 3 of an item with 10 in stock showed 5 packable instead of 2.

Group the bundle components by item and sum their qty before dividing.

* test(selling): cover a repeated component in Available Stock for Packing Items

erpnext/accounts/doctype/purchase_invoice/purchase_invoice.py

@@ -40,6 +40,7 @@ from erpnext.accounts.utils import (
 )
 from erpnext.assets.doctype.asset.asset import is_cwip_accounting_enabled
 from erpnext.controllers.buying_controller import BuyingController
+from erpnext.stock.doctype.purchase_receipt.services.billing_status import is_billed_by_qty
 
 
 class WarehouseMissingError(frappe.ValidationError):
@@ -296,7 +297,23 @@ class PurchaseInvoice(BuyingController):
 
 		from erpnext.accounts.services.billing_validation import BillingValidationService
 
-		BillingValidationService(self).validate_multiple_billing("Purchase Receipt", "pr_detail", "amount")
+		buying_settings = frappe.get_cached_doc("Buying Settings")
+		billing_validation = BillingValidationService(self)
+		if buying_settings.set_landed_cost_based_on_purchase_invoice_rate:
+			billing_validation.validate_multiple_billing(
+				"Purchase Receipt",
+				"pr_detail",
+				"qty",
+				reference_field="received_qty"
+				if buying_settings.bill_for_rejected_quantity_in_purchase_invoice
+				else "qty",
+				billing_flag="update_billed_amount_in_purchase_receipt",
+			)
+			billing_validation.validate_multiple_billing(
+				"Purchase Order", "po_detail", "qty", billing_flag="update_billed_amount_in_purchase_order"
+			)
+		else:
+			billing_validation.validate_multiple_billing("Purchase Receipt", "pr_detail", "amount")
 		self.set_status()
 		self.validate_purchase_receipt_if_update_stock()
 		self.validate_exchange_rate_with_purchase_receipt()
@@ -589,6 +606,9 @@ class PurchaseInvoice(BuyingController):
 					frappe.throw(_("Purchase Receipt {0} is not submitted").format(d.purchase_receipt))
 
 	def update_status_updater_args(self):
+		if is_billed_by_qty():
+			self.set_purchase_order_billing_by_qty()
+
 		if cint(self.update_stock):
 			self.status_updater.append(
 				{
@@ -639,6 +659,13 @@ class PurchaseInvoice(BuyingController):
 					}
 				)
 
+	def set_purchase_order_billing_by_qty(self):
+		"""The status updater keeps billed_amt current; billing % and over-billing follow invoiced qty instead."""
+		for args in self.status_updater:
+			if args.get("overflow_type") == "billing":
+				args.pop("percent_join_field", None)
+				args["validate_overflow"] = False
+
 	def validate_purchase_receipt_if_update_stock(self):
 		if self.update_stock:
 			for item in self.get("items"):
@@ -672,6 +699,7 @@ class PurchaseInvoice(BuyingController):
 
 		self.update_status_updater_args()
 		self.update_prevdoc_status()
+		BillingStatusService(self).update_billing_status_in_po()
 
 		frappe.get_cached_doc("Authorization Control").validate_approving_authority(
 			self.doctype, self.company, self.base_grand_total
@@ -786,6 +814,7 @@ class PurchaseInvoice(BuyingController):
 
 		self.update_status_updater_args()
 		self.update_prevdoc_status()
+		BillingStatusService(self).update_billing_status_in_po()
 
 		if not self.is_return:
 			self.update_billing_status_for_zero_amount_refdoc("Purchase Receipt")

erpnext/accounts/doctype/purchase_invoice/services/billing_status.py

@@ -1,7 +1,7 @@
 # Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
 # License: GNU General Public License v3. See license.txt
 
-"""Purchase Receipt billing sync and provisional-entry cancellation for Purchase Invoice."""
+"""Purchase Receipt and Purchase Order billing sync and provisional-entry cancellation for Purchase Invoice."""
 
 import frappe
 from frappe import qb
@@ -9,6 +9,9 @@ from frappe.query_builder.functions import Sum
 from frappe.utils import flt
 
 from erpnext.stock.doctype.purchase_receipt.services.billing_status import (
+	get_purchase_receipts_against_po_details,
+	get_receipt_billing_data,
+	is_billed_by_qty,
 	update_billed_amount_based_on_po,
 	update_billing_percentage,
 )
@@ -44,16 +47,53 @@ class BillingStatusService:
 		if po_details:
 			updated_pr += update_billed_amount_based_on_po(po_details, update_modified)
 
-		adjust_incoming_rate = frappe.db.get_single_value(
-			"Buying Settings", "set_landed_cost_based_on_purchase_invoice_rate"
-		)
+		if not is_billed_by_qty():
+			for pr in set(updated_pr):
+				pr_doc = frappe.get_lazy_doc("Purchase Receipt", pr)
+				update_billing_percentage(pr_doc, update_modified=update_modified)
+			return
 
-		for pr in set(updated_pr):
-			pr_doc = frappe.get_lazy_doc("Purchase Receipt", pr)
+		self.update_billing_status_in_receipts_on_po_lines(set(updated_pr), update_modified)
+
+	def update_billing_status_in_receipts_on_po_lines(self, updated_pr: set, update_modified: bool) -> None:
+		"""Order invoices are spread over every receipt on the line, so all of them are refreshed from one split."""
+		pr_docs = [
+			frappe.get_lazy_doc("Purchase Receipt", pr) for pr in updated_pr | self.get_receipts_on_po_lines()
+		]
+
+		pr_items = []
+		for pr_doc in pr_docs:
+			pr_items.extend(pr_doc.items)
+
+		bill_for_rejected = frappe.db.get_single_value(
+			"Buying Settings", "bill_for_rejected_quantity_in_purchase_invoice"
+		)
+		billing_data = get_receipt_billing_data(pr_items, bill_for_rejected)
+
+		for pr_doc in pr_docs:
 			update_billing_percentage(
-				pr_doc, update_modified=update_modified, adjust_incoming_rate=adjust_incoming_rate
+				pr_doc,
+				update_modified=update_modified,
+				adjust_incoming_rate=True,
+				billing_data=billing_data,
+				is_refresh=True,
 			)
 
+	def get_receipts_on_po_lines(self) -> set:
+		po_details = list({d.po_detail for d in self.doc.get("items") if d.po_detail})
+		if not po_details:
+			return set()
+
+		return {pr_item.parent for pr_item in get_purchase_receipts_against_po_details(po_details)}
+
+	def update_billing_status_in_po(self) -> None:
+		doc = self.doc
+		if not is_billed_by_qty() or (doc.is_return and not doc.update_billed_amount_in_purchase_order):
+			return
+
+		for purchase_order in {item.purchase_order for item in doc.items if item.purchase_order}:
+			frappe.get_doc("Purchase Order", purchase_order).update_billing_percentage()
+
 	def get_pr_details_billed_amt(self) -> dict:
 		# Get billed amount based on purchase receipt item reference (pr_detail) in purchase invoice
 

erpnext/accounts/doctype/purchase_invoice/test_purchase_invoice.py

@@ -3787,6 +3787,74 @@ class TestPurchaseInvoice(ERPNextTestSuite, StockTestMixin):
 		# Test 4 - Since this PI is overbilled by 130% and only 120% is allowed, it will fail
 		self.assertRaises(frappe.ValidationError, pi.submit)
 
+	@ERPNextTestSuite.change_settings("Accounts Settings", {"over_billing_allowance": 0})
+	@ERPNextTestSuite.change_settings(
+		"Buying Settings",
+		{
+			"maintain_same_rate": 0,
+			"set_landed_cost_based_on_purchase_invoice_rate": 1,
+			"bill_for_rejected_quantity_in_purchase_invoice": 0,
+		},
+	)
+	def test_receipt_over_billing_by_qty_when_landed_cost_follows_invoice_rate(self):
+		pr = make_purchase_receipt(qty=100, rate=50)
+		for qty in (25, 75):
+			pi = create_purchase_invoice_from_receipt(pr.name)
+			pi.items[0].qty = qty
+			pi.items[0].rate = 200
+			pi.submit()
+
+		pr.reload()
+		self.assertEqual(pr.status, "Completed")
+
+		extra_invoice = frappe.copy_doc(pi)
+		extra_invoice.items[0].qty = 100
+		self.assertRaisesRegex(frappe.ValidationError, "Cannot overbill", extra_invoice.submit)
+
+	@ERPNextTestSuite.change_settings("Accounts Settings", {"over_billing_allowance": 0})
+	@ERPNextTestSuite.change_settings(
+		"Buying Settings",
+		{
+			"maintain_same_rate": 0,
+			"set_landed_cost_based_on_purchase_invoice_rate": 1,
+			"bill_for_rejected_quantity_in_purchase_invoice": 1,
+		},
+	)
+	def test_qty_over_billing_counts_billed_rejected_qty(self):
+		pr = make_purchase_receipt(received_qty=100, qty=90, rejected_qty=10, rate=50)
+		pi = create_purchase_invoice_from_receipt(pr.name)
+		pi.submit()
+		self.assertEqual(pi.items[0].qty, 100)
+
+		extra_invoice = frappe.copy_doc(pi)
+		extra_invoice.items[0].qty = 50
+		self.assertRaisesRegex(frappe.ValidationError, "Cannot overbill", extra_invoice.submit)
+
+	@ERPNextTestSuite.change_settings("Accounts Settings", {"over_billing_allowance": 0})
+	@ERPNextTestSuite.change_settings(
+		"Buying Settings",
+		{
+			"maintain_same_rate": 0,
+			"set_landed_cost_based_on_purchase_invoice_rate": 1,
+			"bill_for_rejected_quantity_in_purchase_invoice": 0,
+		},
+	)
+	def test_non_updating_debit_note_gives_no_qty_billing_room(self):
+		from erpnext.accounts.doctype.purchase_invoice.mapper import make_debit_note
+
+		pr = make_purchase_receipt(qty=100, rate=50)
+		pi = create_purchase_invoice_from_receipt(pr.name)
+		pi.submit()
+
+		debit_note = make_debit_note(pi.name)
+		debit_note.items[0].qty = -20
+		debit_note.update_billed_amount_in_purchase_receipt = 0
+		debit_note.submit()
+
+		extra_invoice = frappe.copy_doc(pi)
+		extra_invoice.items[0].qty = 20
+		self.assertRaisesRegex(frappe.ValidationError, "Cannot overbill", extra_invoice.submit)
+
 	@ERPNextTestSuite.change_settings("Accounts Settings", {"over_billing_allowance": 0})
 	def test_non_stock_item_over_billing_against_po_is_blocked(self):
 		service_item = create_item(

erpnext/accounts/services/billing_validation.py

@@ -13,10 +13,24 @@ class BillingValidationService:
 	def __init__(self, doc):
 		self.doc = doc
 
-	def validate_multiple_billing(self, ref_dt: str, item_ref_dn: str, based_on: str) -> None:
+	def validate_multiple_billing(
+		self,
+		ref_dt: str,
+		item_ref_dn: str,
+		based_on: str,
+		reference_field: str | None = None,
+		billing_flag: str | None = None,
+	) -> None:
+		"""`reference_field` is the reference row's field to bill against, `based_on` by default.
+		With `billing_flag`, debit notes that have that invoice field off do not count as billing."""
 		from erpnext.controllers.status_updater import get_allowance_for
 
-		ref_wise_billed_amount = self.get_reference_wise_billed_amt(ref_dt, item_ref_dn, based_on)
+		if billing_flag and self.doc.get("is_return") and not self.doc.get(billing_flag):
+			return
+
+		ref_wise_billed_amount = self.get_reference_wise_billed_amt(
+			ref_dt, item_ref_dn, based_on, reference_field or based_on, billing_flag
+		)
 		if not ref_wise_billed_amount:
 			return
 
@@ -41,15 +55,11 @@ class BillingValidationService:
 			total_overbilled_amt += overbill_amt
 
 			if overbill_amt > precision_allowance and not is_overbilling_allowed:
-				if self.doc.doctype != "Purchase Invoice" or not cint(
-					frappe.db.get_single_value(
-						"Buying Settings", "bill_for_rejected_quantity_in_purchase_invoice"
-					)
-				):
+				if not self.is_rejected_qty_billed_by_amount(based_on):
 					overbilled_items.append(row)
 
 		if overbilled_items:
-			self.throw_overbill_exception(overbilled_items, precision)
+			self.throw_overbill_exception(overbilled_items, precision, based_on)
 
 		if is_overbilling_allowed and total_overbilled_amt > 0.1:
 			frappe.msgprint(
@@ -60,15 +70,31 @@ class BillingValidationService:
 				alert=True,
 			)
 
-	def get_reference_wise_billed_amt(self, ref_dt: str, item_ref_dn: str, based_on: str) -> dict | None:
+	def is_rejected_qty_billed_by_amount(self, based_on: str) -> bool:
+		"""Billed rejected qty has no value on the receipt, so an amount check cannot hold it."""
+		return (
+			based_on == "amount"
+			and self.doc.doctype == "Purchase Invoice"
+			and cint(
+				frappe.db.get_single_value(
+					"Buying Settings", "bill_for_rejected_quantity_in_purchase_invoice"
+				)
+			)
+		)
+
+	def get_reference_wise_billed_amt(
+		self, ref_dt: str, item_ref_dn: str, based_on: str, reference_field: str, billing_flag: str | None
+	) -> dict | None:
 		"""Return sum of billed amounts per reference row, including previously submitted invoices."""
 		reference_names = [d.get(item_ref_dn) for d in self.doc.items if d.get(item_ref_dn)]
 		if not reference_names:
 			return
 
 		precision = self.doc.precision(based_on, "items")
-		reference_details = self.get_billing_reference_details(reference_names, ref_dt + " Item", based_on)
-		already_billed = self.get_already_billed_amount(reference_names, item_ref_dn, based_on)
+		reference_details = self.get_billing_reference_details(
+			reference_names, ref_dt + " Item", reference_field
+		)
+		already_billed = self.get_already_billed_amount(reference_names, item_ref_dn, based_on, billing_flag)
 
 		ref_wise_billed_amount = {}
 		for item in self.doc.items:
@@ -92,7 +118,9 @@ class BillingValidationService:
 
 			ref_wise_billed_amount.setdefault(
 				key,
-				frappe._dict(item_code=item.item_code, billed_amt=0.0, ref_amt=ref_amt, rows=[]),
+				frappe._dict(
+					item_code=item.item_code, uom=item.get("uom"), billed_amt=0.0, ref_amt=ref_amt, rows=[]
+				),
 			)
 			ref_wise_billed_amount[key]["rows"].append(item.idx)
 			ref_wise_billed_amount[key]["ref_amt"] = ref_amt
@@ -115,23 +143,31 @@ class BillingValidationService:
 		)
 
 	def get_already_billed_amount(
-		self, reference_names: list, item_ref_dn: str, based_on: str
+		self, reference_names: list, item_ref_dn: str, based_on: str, billing_flag: str | None = None
 	) -> frappe._dict:
 		item_doctype = frappe.qb.DocType(self.doc.items[0].doctype)
-		based_on_field = frappe.qb.Field(based_on)
-		join_field = frappe.qb.Field(item_ref_dn)
+		based_on_field = item_doctype[based_on]
+		join_field = item_doctype[item_ref_dn]
 
-		return frappe._dict(
-			(
-				frappe.qb.from_(item_doctype)
-				.select(join_field, Sum(based_on_field))
-				.where(join_field.isin(reference_names))
-				.where((item_doctype.docstatus == 1) & (item_doctype.parent != self.doc.name))
-				.groupby(join_field)
-			).run()
+		query = (
+			frappe.qb.from_(item_doctype)
+			.select(join_field, Sum(based_on_field))
+			.where(join_field.isin(reference_names))
+			.where((item_doctype.docstatus == 1) & (item_doctype.parent != self.doc.name))
+			.groupby(join_field)
 		)
 
-	def throw_overbill_exception(self, overbilled_items: list, precision: int) -> None:
+		if billing_flag:
+			invoice = frappe.qb.DocType(self.doc.doctype)
+			query = (
+				query.inner_join(invoice)
+				.on(invoice.name == item_doctype.parent)
+				.where((invoice.is_return == 0) | (invoice[billing_flag] == 1))
+			)
+
+		return frappe._dict(query.run())
+
+	def throw_overbill_exception(self, overbilled_items: list, precision: int, based_on: str) -> None:
 		message = (
 			_("<p>Cannot overbill for the following Items:</p>")
 			+ "<ul>"
@@ -139,9 +175,7 @@ class BillingValidationService:
 				_("<li>Item {0} in row(s) {1} billed more than {2}</li>").format(
 					frappe.bold(item.item_code),
 					", ".join(str(x) for x in item.rows),
-					frappe.bold(
-						fmt_money(item.max_allowed_amt, precision=precision, currency=self.doc.currency)
-					),
+					frappe.bold(self.get_formatted_limit(item, precision, based_on)),
 				)
 				for item in overbilled_items
 			)
@@ -149,3 +183,9 @@ class BillingValidationService:
 		)
 		message += _("<p>To allow over-billing, please set allowance in Accounts Settings.</p>")
 		frappe.throw(message)
+
+	def get_formatted_limit(self, item: frappe._dict, precision: int, based_on: str) -> str:
+		if based_on == "qty":
+			return f"{flt(item.max_allowed_amt, precision)} {item.uom}"
+
+		return fmt_money(item.max_allowed_amt, precision=precision, currency=self.doc.currency)

erpnext/buying/doctype/purchase_order/services/status.py

@@ -10,6 +10,10 @@ from frappe.utils import cstr, flt
 
 from erpnext.buying.doctype.purchase_order.services.subcontracting import SubcontractingService
 from erpnext.controllers.item_close import validate_parent_reopen
+from erpnext.stock.doctype.purchase_receipt.services.billing_status import (
+	get_invoiced_qty_against_po_items,
+	get_qty_based_percent_billed,
+)
 
 
 class StatusService:
@@ -61,3 +65,9 @@ class StatusService:
 
 		per_received = flt(received_qty / total_qty) * 100 if total_qty else 0
 		doc.db_set("per_received", per_received, update_modified=False)
+
+	def get_percent_billed_by_qty(self) -> float:
+		items = [item for item in self.doc.items if not item.closed] or self.doc.items
+		billable_qty = {item.name: flt(item.qty) for item in items}
+		invoiced_qty = get_invoiced_qty_against_po_items(list(billable_qty))
+		return get_qty_based_percent_billed(items, billable_qty, invoiced_qty)

erpnext/buying/doctype/purchase_order/test_purchase_order.py

@@ -1521,6 +1521,90 @@ class TestPurchaseOrder(ERPNextTestSuite):
 		po.reload()
 		self.assertEqual(po.per_billed, 100)
 
+	@ERPNextTestSuite.change_settings(
+		"Buying Settings", {"maintain_same_rate": 0, "set_landed_cost_based_on_purchase_invoice_rate": 1}
+	)
+	def test_per_billed_by_qty_when_landed_cost_follows_invoice_rate(self):
+		po = create_purchase_order(qty=100, rate=50)
+		pi = make_pi_from_po(po.name)
+		pi.items[0].qty = 25
+		pi.items[0].rate = 200
+		pi.submit()
+
+		po.reload()
+		self.assertEqual(po.per_billed, 25)
+		self.assertEqual(po.status, "To Receive and Bill")
+
+		pi.reload()
+		pi.cancel()
+		po.reload()
+		self.assertEqual(po.per_billed, 0)
+
+	@ERPNextTestSuite.change_settings("Accounts Settings", {"over_billing_allowance": 0})
+	@ERPNextTestSuite.change_settings(
+		"Buying Settings",
+		{
+			"maintain_same_rate": 0,
+			"set_landed_cost_based_on_purchase_invoice_rate": 1,
+			"bill_for_rejected_quantity_in_purchase_invoice": 0,
+		},
+	)
+	def test_over_billing_by_qty_when_landed_cost_follows_invoice_rate(self):
+		po = create_purchase_order(qty=100, rate=50)
+		for qty in (25, 75):
+			pi = make_pi_from_po(po.name)
+			pi.items[0].qty = qty
+			pi.items[0].rate = 200
+			pi.submit()
+
+		po.reload()
+		self.assertEqual(po.per_billed, 100)
+
+		extra_invoice = frappe.copy_doc(pi)
+		extra_invoice.items[0].qty = 100
+		self.assertRaisesRegex(frappe.ValidationError, "Cannot overbill", extra_invoice.submit)
+
+	@ERPNextTestSuite.change_settings("Accounts Settings", {"over_billing_allowance": 0})
+	@ERPNextTestSuite.change_settings(
+		"Buying Settings",
+		{
+			"maintain_same_rate": 0,
+			"set_landed_cost_based_on_purchase_invoice_rate": 1,
+			"bill_for_rejected_quantity_in_purchase_invoice": 1,
+		},
+	)
+	def test_qty_over_billing_holds_with_bill_for_rejected(self):
+		po = create_purchase_order(qty=100, rate=50)
+		pi = make_pi_from_po(po.name)
+		pi.submit()
+
+		self.assertRaisesRegex(frappe.ValidationError, "Cannot overbill", frappe.copy_doc(pi).submit)
+
+	@ERPNextTestSuite.change_settings("Accounts Settings", {"over_billing_allowance": 0})
+	@ERPNextTestSuite.change_settings(
+		"Buying Settings",
+		{
+			"maintain_same_rate": 0,
+			"set_landed_cost_based_on_purchase_invoice_rate": 1,
+			"bill_for_rejected_quantity_in_purchase_invoice": 0,
+		},
+	)
+	def test_non_updating_debit_note_gives_no_order_billing_room(self):
+		from erpnext.accounts.doctype.purchase_invoice.mapper import make_debit_note
+
+		po = create_purchase_order(qty=100, rate=50)
+		pi = make_pi_from_po(po.name)
+		pi.submit()
+
+		debit_note = make_debit_note(pi.name)
+		debit_note.items[0].qty = -20
+		debit_note.update_billed_amount_in_purchase_order = 0
+		debit_note.submit()
+
+		extra_invoice = frappe.copy_doc(pi)
+		extra_invoice.items[0].qty = 20
+		self.assertRaisesRegex(frappe.ValidationError, "Cannot overbill", extra_invoice.submit)
+
 	@ERPNextTestSuite.change_settings("Buying Settings", {"allow_zero_qty_in_purchase_order": 1})
 	def test_receive_zero_qty_purchase_order(self):
 		"""

erpnext/buying/report/purchase_order_analysis/purchase_order_analysis.js

@@ -90,6 +90,30 @@ frappe.query_reports["Purchase Order Analysis"] = {
 			label: __("Group by Purchase Order"),
 			fieldtype: "Check",
 			default: 0,
+			on_change: (report) => {
+				if (report.get_filter_value("group_by_po") && report.get_filter_value("group_by_item")) {
+					report.set_filter_value("group_by_item", 0);
+					return;
+				}
+				if (!report._no_refresh) {
+					report.refresh(true);
+				}
+			},
+		},
+		{
+			fieldname: "group_by_item",
+			label: __("Group by Item"),
+			fieldtype: "Check",
+			default: 0,
+			on_change: (report) => {
+				if (report.get_filter_value("group_by_po") && report.get_filter_value("group_by_item")) {
+					report.set_filter_value("group_by_po", 0);
+					return;
+				}
+				if (!report._no_refresh) {
+					report.refresh(true);
+				}
+			},
 		},
 	],
 

erpnext/buying/report/purchase_order_analysis/purchase_order_analysis.py

@@ -9,11 +9,16 @@ from frappe import _
 from frappe.query_builder.functions import IfNull, Sum
 from frappe.utils import date_diff, flt, getdate
 
+import erpnext
+
 
 def execute(filters=None):
 	if not filters:
 		return [], []
 
+	filters = frappe._dict(filters)
+	filters.company = filters.get("company") or erpnext.get_default_company()
+
 	validate_filters(filters)
 
 	columns = get_columns(filters)
@@ -30,6 +35,9 @@ def execute(filters=None):
 
 
 def validate_filters(filters):
+	if not filters.get("company"):
+		frappe.throw(_("{0} is mandatory").format(_("Company")))
+
 	from_date, to_date = filters.get("from_date"), filters.get("to_date")
 
 	if not from_date and to_date:
@@ -37,6 +45,9 @@ def validate_filters(filters):
 	elif date_diff(to_date, from_date) < 0:
 		frappe.throw(_("To Date cannot be before From Date."))
 
+	if filters.get("group_by_po") and filters.get("group_by_item"):
+		frappe.throw(_("Group the report by Purchase Order or by Item, not both."))
+
 
 def get_data(filters):
 	po = frappe.qb.DocType("Purchase Order")
@@ -57,6 +68,7 @@ def get_data(filters):
 			po.status,
 			po.supplier,
 			po_item.item_code,
+			po_item.uom,
 			po_item.qty,
 			po_item.received_qty,
 			(po_item.qty - po_item.received_qty).as_("pending_qty"),
@@ -71,15 +83,13 @@ def get_data(filters):
 			po_item.name,
 		)
 		.where((po_item.parent == po.name) & (po.status.notin(("Stopped", "On Hold"))) & (po.docstatus == 1))
+		.where(po.company == filters.get("company"))
 		# the selected po.* columns need the Purchase Order PK grouped on postgres; po.name is 1:1
 		# with the grouped po_item.name, so groups are unchanged.
 		.groupby(po_item.name, po.name)
 		.orderby(po.transaction_date)
 	)
 
-	if filters.get("company"):
-		query = query.where(po.company == filters.get("company"))
-
 	if filters.get("name"):
 		query = query.where(po.name.isin(filters.get("name")))
 
@@ -133,60 +143,78 @@ def get_received_amount_data(data):
 	return frappe._dict(data)
 
 
+AGGREGATED_FIELDS = (
+	"qty",
+	"received_qty",
+	"pending_qty",
+	"billed_qty",
+	"qty_to_bill",
+	"amount",
+	"received_qty_amount",
+	"billed_amount",
+	"pending_amount",
+)
+
+
 def prepare_data(data, filters):
 	completed, pending = 0, 0
-	pending_field = "pending_amount"
-	completed_field = "billed_amount"
-
-	if filters.get("group_by_po"):
-		purchase_order_map = {}
 
 	for row in data:
-		# sum data for chart
-		completed += row[completed_field]
-		pending += row[pending_field]
+		completed += row["billed_amount"]
+		pending += row["pending_amount"]
 
-		# prepare data for report view
 		row["qty_to_bill"] = flt(row["qty"]) - flt(row["billed_qty"])
 
-		if filters.get("group_by_po"):
-			po_name = row["purchase_order"]
-
-			if po_name not in purchase_order_map:
-				# create an entry
-				row_copy = copy.deepcopy(row)
-				purchase_order_map[po_name] = row_copy
-			else:
-				# update existing entry
-				po_row = purchase_order_map[po_name]
-				po_row["required_date"] = min(getdate(po_row["required_date"]), getdate(row["required_date"]))
-
-				# sum numeric columns
-				fields = [
-					"qty",
-					"received_qty",
-					"pending_qty",
-					"billed_qty",
-					"qty_to_bill",
-					"amount",
-					"received_qty_amount",
-					"billed_amount",
-					"pending_amount",
-				]
-				for field in fields:
-					po_row[field] = flt(row[field]) + flt(po_row[field])
-
 	chart_data = prepare_chart_data(pending, completed)
 
 	if filters.get("group_by_po"):
-		data = []
-		for po in purchase_order_map:
-			data.append(purchase_order_map[po])
-		return data, chart_data
+		data = group_by_purchase_order(data)
+	elif filters.get("group_by_item"):
+		data = group_by_item(data)
 
 	return data, chart_data
 
 
+def group_by_purchase_order(data):
+	purchase_order_map = {}
+
+	for row in data:
+		group = purchase_order_map.get(row["purchase_order"])
+		if not group:
+			purchase_order_map[row["purchase_order"]] = copy.deepcopy(row)
+			continue
+
+		group["required_date"] = min(getdate(group["required_date"]), getdate(row["required_date"]))
+		add_aggregated_fields(group, row)
+
+	return list(purchase_order_map.values())
+
+
+def group_by_item(data):
+	"""Group on company and UOM as well as the item.
+
+	Quantities are in the line UOM and amounts are in the company currency, so neither sums
+	across a second UOM of the same item or a second company.
+	"""
+	item_map = {}
+
+	for row in data:
+		key = (row["company"], row["item_code"], row["uom"])
+		group = item_map.get(key)
+		if not group:
+			item_map[key] = copy.deepcopy(row)
+			continue
+
+		add_aggregated_fields(group, row)
+
+	return sorted(item_map.values(), key=lambda row: (row["company"], row["item_code"], row["uom"]))
+
+
+def add_aggregated_fields(group, row):
+	for field in AGGREGATED_FIELDS:
+		group[field] = flt(group[field]) + flt(row[field])
+
+
 def prepare_chart_data(pending, completed):
 	labels = [_("Amount to Bill"), _("Billed Amount")]
 
@@ -198,7 +226,30 @@ def prepare_chart_data(pending, completed):
 
 
 def get_columns(filters):
-	columns = [
+	if filters.get("group_by_item"):
+		return get_grouped_by_item_columns()
+
+	columns = get_purchase_order_columns()
+
+	if not filters.get("group_by_po"):
+		columns.append(get_item_code_column())
+
+	columns += get_quantity_columns() + get_amount_columns()
+	columns += [get_warehouse_column(), get_company_column()]
+
+	return columns
+
+
+def get_grouped_by_item_columns():
+	columns = [get_item_code_column(), get_uom_column()]
+	columns += get_quantity_columns() + get_amount_columns()
+	columns.append(get_company_column())
+
+	return columns
+
+
+def get_purchase_order_columns():
+	return [
 		{"label": _("Date"), "fieldname": "date", "fieldtype": "Date", "width": 90},
 		{"label": _("Required By"), "fieldname": "required_date", "fieldtype": "Date", "width": 90},
 		{
@@ -225,101 +276,119 @@ def get_columns(filters):
 		},
 	]
 
-	if not filters.get("group_by_po"):
-		columns.append(
-			{
-				"label": _("Item Code"),
-				"fieldname": "item_code",
-				"fieldtype": "Link",
-				"options": "Item",
-				"width": 100,
-			}
-		)
 
-	columns.extend(
-		[
-			{
-				"label": _("Qty"),
-				"fieldname": "qty",
-				"fieldtype": "Float",
-				"width": 120,
-				"convertible": "qty",
-			},
-			{
-				"label": _("Received Qty"),
-				"fieldname": "received_qty",
-				"fieldtype": "Float",
-				"width": 120,
-				"convertible": "qty",
-			},
-			{
-				"label": _("Pending Qty"),
-				"fieldname": "pending_qty",
-				"fieldtype": "Float",
-				"width": 80,
-				"convertible": "qty",
-			},
-			{
-				"label": _("Billed Qty"),
-				"fieldname": "billed_qty",
-				"fieldtype": "Float",
-				"width": 80,
-				"convertible": "qty",
-			},
-			{
-				"label": _("Qty to Bill"),
-				"fieldname": "qty_to_bill",
-				"fieldtype": "Float",
-				"width": 80,
-				"convertible": "qty",
-			},
-			{
-				"label": _("Amount"),
-				"fieldname": "amount",
-				"fieldtype": "Currency",
-				"width": 110,
-				"options": "Company:company:default_currency",
-				"convertible": "rate",
-			},
-			{
-				"label": _("Billed Amount"),
-				"fieldname": "billed_amount",
-				"fieldtype": "Currency",
-				"width": 110,
-				"options": "Company:company:default_currency",
-				"convertible": "rate",
-			},
-			{
-				"label": _("Pending Amount"),
-				"fieldname": "pending_amount",
-				"fieldtype": "Currency",
-				"width": 130,
-				"options": "Company:company:default_currency",
-				"convertible": "rate",
-			},
-			{
-				"label": _("Received Qty Amount"),
-				"fieldname": "received_qty_amount",
-				"fieldtype": "Currency",
-				"width": 130,
-				"options": "Company:company:default_currency",
-				"convertible": "rate",
-			},
-			{
-				"label": _("Warehouse"),
-				"fieldname": "warehouse",
-				"fieldtype": "Link",
-				"options": "Warehouse",
-				"width": 100,
-			},
-			{
-				"label": _("Company"),
-				"fieldname": "company",
-				"fieldtype": "Link",
-				"options": "Company",
-				"width": 100,
-			},
-		]
-	)
+def get_item_code_column():
+	return {
+		"label": _("Item Code"),
+		"fieldname": "item_code",
+		"fieldtype": "Link",
+		"options": "Item",
+		"width": 100,
+	}
 
-	return columns
+
+def get_uom_column():
+	return {
+		"label": _("UOM"),
+		"fieldname": "uom",
+		"fieldtype": "Link",
+		"options": "UOM",
+		"width": 100,
+	}
+
+
+def get_quantity_columns():
+	return [
+		{
+			"label": _("Qty"),
+			"fieldname": "qty",
+			"fieldtype": "Float",
+			"width": 120,
+			"convertible": "qty",
+		},
+		{
+			"label": _("Received Qty"),
+			"fieldname": "received_qty",
+			"fieldtype": "Float",
+			"width": 120,
+			"convertible": "qty",
+		},
+		{
+			"label": _("Pending Qty"),
+			"fieldname": "pending_qty",
+			"fieldtype": "Float",
+			"width": 80,
+			"convertible": "qty",
+		},
+		{
+			"label": _("Billed Qty"),
+			"fieldname": "billed_qty",
+			"fieldtype": "Float",
+			"width": 80,
+			"convertible": "qty",
+		},
+		{
+			"label": _("Qty to Bill"),
+			"fieldname": "qty_to_bill",
+			"fieldtype": "Float",
+			"width": 80,
+			"convertible": "qty",
+		},
+	]
+
+
+def get_amount_columns():
+	return [
+		{
+			"label": _("Amount"),
+			"fieldname": "amount",
+			"fieldtype": "Currency",
+			"width": 110,
+			"options": "Company:company:default_currency",
+			"convertible": "rate",
+		},
+		{
+			"label": _("Billed Amount"),
+			"fieldname": "billed_amount",
+			"fieldtype": "Currency",
+			"width": 110,
+			"options": "Company:company:default_currency",
+			"convertible": "rate",
+		},
+		{
+			"label": _("Pending Amount"),
+			"fieldname": "pending_amount",
+			"fieldtype": "Currency",
+			"width": 130,
+			"options": "Company:company:default_currency",
+			"convertible": "rate",
+		},
+		{
+			"label": _("Received Qty Amount"),
+			"fieldname": "received_qty_amount",
+			"fieldtype": "Currency",
+			"width": 130,
+			"options": "Company:company:default_currency",
+			"convertible": "rate",
+		},
+	]
+
+
+def get_warehouse_column():
+	return {
+		"label": _("Warehouse"),
+		"fieldname": "warehouse",
+		"fieldtype": "Link",
+		"options": "Warehouse",
+		"width": 100,
+	}
+
+
+def get_company_column():
+	return {
+		"label": _("Company"),
+		"fieldname": "company",
+		"fieldtype": "Link",
+		"options": "Company",
+		"width": 100,
+	}

erpnext/buying/report/purchase_order_analysis/test_purchase_order_analysis.py

@@ -1,28 +1,150 @@
 # Copyright (c) 2025, Frappe Technologies Pvt. Ltd. and contributors
 # For license information, please see license.txt
 
+from unittest.mock import patch
+
+import frappe
 from frappe.utils import add_days, nowdate
 
+from erpnext.buying.doctype.purchase_order.mapper import make_purchase_invoice
+from erpnext.buying.doctype.purchase_order.test_purchase_order import (
+	create_pr_against_po,
+	create_purchase_order,
+)
+from erpnext.buying.report.purchase_order_analysis.purchase_order_analysis import (
+	AGGREGATED_FIELDS,
+	execute,
+	group_by_item,
+)
+from erpnext.stock.doctype.item.test_item import create_item
 from erpnext.tests.utils import ERPNextTestSuite
 
+ITEM_CODE = "_Test PO Analysis Item"
+
 
 class TestPurchaseOrderAnalysis(ERPNextTestSuite):
+	def get_filters(self, **filters):
+		return {
+			"company": "_Test Company",
+			"from_date": add_days(nowdate(), -1),
+			"to_date": add_days(nowdate(), 1),
+			**filters,
+		}
+
+	def make_purchase_order(self, qty, uom=None):
+		create_item(ITEM_CODE)
+		po = create_purchase_order(item_code=ITEM_CODE, qty=qty, do_not_save=True)
+		if uom:
+			po.items[0].uom = uom
+		po.set_missing_values()
+		po.insert()
+		po.submit()
+		return po
+
+	def add_uom(self, uom, conversion_factor):
+		item = frappe.get_doc("Item", ITEM_CODE)
+		if not any(row.uom == uom for row in item.uoms):
+			item.append("uoms", {"uom": uom, "conversion_factor": conversion_factor})
+			item.save()
+
+	def make_item_row(self, company, qty):
+		row = frappe._dict(dict.fromkeys(AGGREGATED_FIELDS, 0))
+		row.update({"company": company, "item_code": ITEM_CODE, "uom": "Nos", "qty": qty})
+		return row
+
+	def get_item_rows(self, data):
+		return [row for row in data if row["item_code"] == ITEM_CODE]
+
 	def test_report_executes_and_lists_po(self):
 		# get_data groups by (Purchase Order Item, Purchase Order) while selecting other parent
 		# columns; this exercises that GROUP BY so the report stays valid on Postgres (which rejects
 		# selecting non-grouped columns).
-		from erpnext.buying.doctype.purchase_order.test_purchase_order import create_purchase_order
-		from erpnext.buying.report.purchase_order_analysis.purchase_order_analysis import execute
-
 		po = create_purchase_order(company="_Test Company")
 
-		filters = {
-			"company": "_Test Company",
-			"from_date": add_days(nowdate(), -1),
-			"to_date": add_days(nowdate(), 1),
-		}
-		result = execute(filters)
+		result = execute(self.get_filters())
 		columns, data = result[0], result[1]
 
 		self.assertTrue(columns)
 		self.assertIn(po.name, {row.get("purchase_order") for row in data})
+
+	def test_group_by_item_across_purchase_orders(self):
+		po = self.make_purchase_order(qty=10)
+		billed_po = self.make_purchase_order(qty=4)
+		create_pr_against_po(po.name, received_qty=3)
+
+		pi = make_purchase_invoice(billed_po.name)
+		pi.items[0].qty = 2
+		pi.insert().submit()
+
+		columns, data, message, chart = execute(self.get_filters(group_by_item=1))
+
+		expected_value = {
+			"uom": "Nos",
+			"qty": 14,
+			"received_qty": 3,
+			"pending_qty": 11,
+			"billed_qty": 2,
+			"qty_to_bill": 12,
+			"amount": 7000,
+			"received_qty_amount": 1500,
+			"billed_amount": 1000,
+			"pending_amount": 6000,
+		}
+		rows = self.get_item_rows(data)
+		self.assertEqual(len(rows), 1)
+		for key, val in expected_value.items():
+			with self.subTest(key=key, val=val):
+				self.assertEqual(rows[0][key], val)
+
+		fieldnames = [column["fieldname"] for column in columns]
+		self.assertIn("uom", fieldnames)
+		self.assertNotIn("purchase_order", fieldnames)
+
+	def test_group_by_item_keeps_each_uom_apart(self):
+		self.make_purchase_order(qty=10)
+		self.add_uom("Box", 10)
+		self.make_purchase_order(qty=2, uom="Box")
+
+		columns, data, message, chart = execute(self.get_filters(group_by_item=1))
+
+		self.assertEqual(
+			[(row["uom"], row["qty"]) for row in self.get_item_rows(data)],
+			[("Box", 2), ("Nos", 10)],
+		)
+
+	def test_group_by_filters_cannot_be_combined(self):
+		self.assertRaises(
+			frappe.ValidationError,
+			execute,
+			self.get_filters(group_by_po=1, group_by_item=1),
+		)
+
+	def test_group_by_item_keeps_each_company_apart(self):
+		rows = [
+			self.make_item_row("_Test Company", 10),
+			self.make_item_row("_Test Company 1", 4),
+			self.make_item_row("_Test Company", 6),
+		]
+
+		self.assertEqual(
+			[(row["company"], row["qty"]) for row in group_by_item(rows)],
+			[("_Test Company", 16), ("_Test Company 1", 4)],
+		)
+
+	def test_company_falls_back_to_the_default(self):
+		po = self.make_purchase_order(qty=10)
+		filters = self.get_filters(company=None)
+
+		with patch("erpnext.get_default_company", return_value="_Test Company"):
+			columns, data, message, chart = execute(filters)
+
+		self.assertIn(po.name, [row["purchase_order"] for row in data])
+
+		with patch("erpnext.get_default_company", return_value="_Test Company 1"):
+			columns, data, message, chart = execute(filters)
+
+		self.assertNotIn(po.name, [row["purchase_order"] for row in data])
+
+	def test_company_is_mandatory_without_a_default(self):
+		with patch("erpnext.get_default_company", return_value=None):
+			self.assertRaises(frappe.ValidationError, execute, self.get_filters(company=None))

erpnext/controllers/status_updater.py

@@ -394,7 +394,7 @@ class StatusUpdater(Document):
 						)
 					)
 
-			if items_to_validate:
+			if items_to_validate and args.get("validate_overflow") is not False:
 				pp_sub_assembly_items = [
 					item.production_plan_sub_assembly_item
 					for item in items_to_validate

erpnext/controllers/stock_controller.py

@@ -33,6 +33,7 @@ from erpnext.setup.doctype.brand.brand import get_brand_defaults
 from erpnext.setup.doctype.item_group.item_group import get_item_group_defaults
 from erpnext.stock import get_warehouse_account, get_warehouse_account_map
 from erpnext.stock.doctype.item.item import get_item_defaults
+from erpnext.stock.doctype.purchase_receipt.services.billing_status import is_billed_by_qty
 from erpnext.stock.services.internal_transfer import StockInternalTransferService
 from erpnext.stock.stock_ledger import get_items_to_be_repost
 
@@ -340,6 +341,10 @@ class StockController(AccountsController):
 		if self.doctype == "Delivery Note":
 			# Bill by amount, falling back to qty when the invoiced amount is short (e.g. rate drop).
 			args["billing_percentage"] = self.get_delivery_note_billing_percentage()
+		elif self.doctype == "Purchase Order" and is_billed_by_qty():
+			from erpnext.buying.doctype.purchase_order.services.status import StatusService
+
+			args["billing_percentage"] = StatusService(self).get_percent_billed_by_qty()
 
 		self._update_percent_field(args, update_modified)
 

erpnext/manufacturing/report/bom_stock_analysis/bom_stock_analysis.py

@@ -326,6 +326,7 @@ def get_producible_fg_items(filters):
 		frappe.throw(_("Warehouse is required to get producible FG Items"))
 
 	bin_subquery = get_stock_qty_by_item(filters).as_("stock_qty")
+	qty_per_unit = Sum(BOM_ITEM.stock_qty) / Max(BOM.quantity)
 
 	query = (
 		frappe.qb.from_(BOM_ITEM)
@@ -339,11 +340,9 @@ def get_producible_fg_items(filters):
 			# item_code -> Max() keeps them valid on postgres with the same value MySQL picked.
 			# description is not: it belongs to the line, so it comes from a representative one below.
 			Max(BOM_ITEM.parent).as_("from_bom_no"),
-			Max(BOM_ITEM.stock_qty / BOM.quantity).as_("qty_per_unit"),
+			qty_per_unit.as_("qty_per_unit"),
 			Max(IfNull(bin_subquery.actual_qty, 0)).as_("available_qty"),
-			Floor(Max(bin_subquery.actual_qty) / ((Sum(BOM_ITEM.stock_qty)) / Max(BOM.quantity))).as_(
-				"producible_qty"
-			),
+			Floor(Max(bin_subquery.actual_qty) / qty_per_unit).as_("producible_qty"),
 		)
 		.where((BOM_ITEM.parent == filters.get("bom")) & (BOM_ITEM.parenttype == "BOM"))
 		.groupby(BOM_ITEM.item_code)

erpnext/manufacturing/report/bom_stock_analysis/test_bom_stock_analysis.py

@@ -7,7 +7,10 @@ from erpnext.manufacturing.doctype.production_plan.test_production_plan import m
 from erpnext.manufacturing.report.bom_stock_analysis.bom_stock_analysis import (
 	execute as bom_stock_analysis_report,
 )
-from erpnext.manufacturing.report.bom_stock_analysis.bom_stock_analysis import get_bom_data
+from erpnext.manufacturing.report.bom_stock_analysis.bom_stock_analysis import (
+	get_bom_data,
+	get_producible_fg_items,
+)
 from erpnext.stock.doctype.item.test_item import make_item
 from erpnext.stock.doctype.stock_reconciliation.test_stock_reconciliation import (
 	create_stock_reconciliation,
@@ -209,6 +212,27 @@ class TestBOMStockAnalysis(ERPNextTestSuite):
 		)
 		self.assertAlmostEqual(flt(rows[0].actual_qty), 10.0, places=6)
 
+	def test_producible_qty_per_unit_sums_repeated_lines(self):
+		"""An item on two BOM lines shows the total per unit, the same basis as the producible qty."""
+		rm = make_item(properties={"is_stock_item": 1, "valuation_rate": 10})
+		fg = make_item(properties={"is_stock_item": 1, "valuation_rate": 10}).name
+
+		bom = make_bom(item=fg, raw_materials=[rm.name], rm_qty=2, do_not_save=True)
+		bom.append(
+			"items",
+			{"item_code": rm.name, "qty": 3, "uom": rm.stock_uom, "stock_uom": rm.stock_uom},
+		)
+		bom.save()
+		bom.submit()
+
+		warehouse = create_warehouse("_Test BOM Stock Analysis Producible")
+		create_stock_reconciliation(item_code=rm.name, warehouse=warehouse, qty=10, rate=10)
+
+		rows = get_producible_fg_items({"bom": bom.name, "warehouse": warehouse})
+		self.assertEqual(len(rows), 1)
+		self.assertEqual(flt(rows[0].qty_per_unit), 5.0)
+		self.assertEqual(flt(rows[0].producible_qty), 2.0)
+
 
 def run_report(bom, warehouse, exploded, qty_to_make):
 	"""Component rows keyed by item code, plus the footer row."""

erpnext/selling/report/available_stock_for_packing_items/available_stock_for_packing_items.py

@@ -3,6 +3,7 @@
 
 
 import frappe
+from frappe.query_builder.functions import Sum
 from frappe.utils import flt
 
 from erpnext.stock.doctype.company_restriction.company_restriction import get_allowed_masters_condition
@@ -77,8 +78,9 @@ def get_item_warehouse_quantity_map():
 		frappe.qb.from_(pbi)
 		.inner_join(pb)
 		.on(pbi.parent == pb.name)
-		.select(pb.new_item_code.as_("parent"), pbi.item_code, pbi.qty)
+		.select(pb.new_item_code.as_("parent"), pbi.item_code, Sum(pbi.qty).as_("qty"))
 		.where((pb.is_active == 1) & (pb.docstatus == 1))
+		.groupby(pb.new_item_code, pbi.item_code)
 	)
 
 	if condition := get_allowed_masters_condition(pb.new_item_code, "Item"):

erpnext/selling/report/available_stock_for_packing_items/test_available_stock_for_packing_items.py

@@ -106,6 +106,18 @@ class TestAvailableStockForPackingItems(ERPNextTestSuite):
 		self.assertEqual(by_warehouse.get(WAREHOUSE), 4.0)
 		self.assertEqual(by_warehouse.get(other_wh), 2.0)
 
+	def test_repeated_component_uses_total_qty(self):
+		comp_a = self.make_component()
+		parent = self.make_bundle_parent()
+
+		self.set_bin_projected_qty(comp_a, WAREHOUSE, 10)
+		self.make_active_bundle(parent, [(comp_a, 2), (comp_a, 3)])
+
+		rows = self.report_rows_for(parent)
+
+		self.assertEqual(len(rows), 1)
+		self.assertEqual(flt(rows[0][5]), 2.0)
+
 	def test_starved_component_drops_row(self):
 		comp_a = self.make_component()
 		comp_b = self.make_component()

erpnext/stock/doctype/purchase_receipt/services/billing_status.py

@@ -115,6 +115,7 @@ def get_purchase_receipts_against_po_details(po_details: list) -> list[dict]:
 		.select(
 			purchase_receipt_item.name,
 			purchase_receipt_item.qty,
+			purchase_receipt_item.rejected_qty,
 			purchase_receipt_item.parent,
 			purchase_receipt_item.amount,
 			purchase_receipt_item.billed_amt,
@@ -127,6 +128,7 @@ def get_purchase_receipts_against_po_details(po_details: list) -> list[dict]:
 		)
 		.orderby(CombineDatetime(purchase_receipt.posting_date, purchase_receipt.posting_time))
 		.orderby(purchase_receipt.name)
+		.orderby(purchase_receipt_item.idx)
 	)
 
 	return query.run(as_dict=True)
@@ -180,30 +182,60 @@ def get_billed_amount_against_po(po_items: list) -> dict:
 
 
 def update_billing_percentage(
-	pr_doc, update_modified: bool = True, adjust_incoming_rate: bool = False
+	pr_doc,
+	update_modified: bool = True,
+	adjust_incoming_rate: bool = False,
+	billing_data: frappe._dict | None = None,
+	is_refresh: bool = False,
 ) -> None:
-	# Update Billing % based on pending accepted qty
+	"""`billing_data` from get_receipt_billing_data lets receipts on one order line share one fetch.
+	A refresh leaves the receipt alone when its billing % did not move."""
 	buying_settings = frappe.get_single("Buying Settings")
+	bill_for_rejected = buying_settings.bill_for_rejected_quantity_in_purchase_invoice
+	items = [item for item in pr_doc.items if not item.closed] or pr_doc.items
+
+	if buying_settings.set_landed_cost_based_on_purchase_invoice_rate:
+		if billing_data is None:
+			billing_data = get_receipt_billing_data(pr_doc.items, bill_for_rejected)
+		percent_billed = get_percent_billed_by_qty(pr_doc, items, bill_for_rejected, billing_data)
+	else:
+		percent_billed = get_percent_billed_by_amount(pr_doc, items, bill_for_rejected)
+
+	precision = pr_doc.precision("per_billed")
+	if not is_refresh or flt(percent_billed, precision) != flt(pr_doc.per_billed, precision):
+		pr_doc.db_set("per_billed", percent_billed)
+		if update_modified:
+			pr_doc.set_status(update=True)
+			pr_doc.notify_update()
+
+	if adjust_incoming_rate and set_amount_difference_with_purchase_invoice(items, billing_data.invoiced):
+		adjust_incoming_rate_for_pr(pr_doc)
+
+
+def get_receipt_billing_data(pr_items: list, bill_for_rejected: bool) -> frappe._dict:
+	"""Invoice split and returned qty for a set of receipt rows, fetched once for all of them."""
+	return frappe._dict(
+		invoiced=get_invoiced_qty_and_amount(pr_items, bill_for_rejected),
+		returned_qty=get_item_wise_returned_qty([item.name for item in pr_items]),
+	)
+
+
+def get_percent_billed_by_amount(pr_doc, items: list, bill_for_rejected: bool) -> float:
 	over_billing_allowance, role_allowed_to_over_bill = frappe.get_single_value(
 		"Accounts Settings", ["over_billing_allowance", "role_allowed_to_over_bill"]
 	)
 
-	total_amount, total_billed_amount, pi_landed_cost_amount = 0, 0, 0
-	item_wise_returned_qty = get_item_wise_returned_qty(pr_doc)
-	billed_qty_amt = frappe._dict()
+	total_amount, total_billed_amount = 0, 0
+	item_wise_returned_qty = get_item_wise_returned_qty([item.name for item in pr_doc.items])
 
-	if adjust_incoming_rate:
-		billed_qty_amt = get_billed_qty_amount_against_purchase_receipt(pr_doc)
-		billed_qty_amt_based_on_po = get_billed_qty_amount_against_purchase_order(pr_doc)
-
-	for item in [item for item in pr_doc.items if not item.closed] or pr_doc.items:
+	for item in items:
 		returned_qty = flt(item_wise_returned_qty.get(item.name))
 		returned_amount = flt(returned_qty) * flt(item.rate)
 		pending_amount = flt(item.amount) - returned_amount
 
 		# When rejected qty is billable, its value is part of the billable base too
 		rejected_amount = 0.0
-		if buying_settings.bill_for_rejected_quantity_in_purchase_invoice:
+		if bill_for_rejected:
 			rejected_amount = flt(item.rejected_qty * item.rate, item.precision("amount"))
 			pending_amount = flt(item.amount) + rejected_amount
 
@@ -219,54 +251,7 @@ def update_billing_percentage(
 
 		amount = flt(item.amount) + rejected_amount
 
-		if adjust_incoming_rate:
-			adjusted_amt = 0.0
-
-			if (
-				item.billed_amt is not None
-				and item.amount is not None
-				and (
-					billed_qty_amt.get(item.name) or billed_qty_amt_based_on_po.get(item.purchase_order_item)
-				)
-			):
-				qty = None
-				if billed_qty_amt.get(item.name):
-					qty = billed_qty_amt.get(item.name).get("qty")
-
-				if not qty and billed_qty_amt_based_on_po.get(item.purchase_order_item):
-					if item.qty < billed_qty_amt_based_on_po.get(item.purchase_order_item)["qty"]:
-						qty = item.qty
-					else:
-						qty = billed_qty_amt_based_on_po.get(item.purchase_order_item)["qty"]
-
-					billed_qty_amt_based_on_po[item.purchase_order_item]["qty"] -= qty
-
-				billed_amt = item.billed_amt
-				if billed_qty_amt.get(item.name):
-					billed_amt = flt(billed_qty_amt.get(item.name).get("amount"))
-				elif billed_qty_amt_based_on_po.get(item.purchase_order_item):
-					total_billed_qty = (
-						billed_qty_amt_based_on_po.get(item.purchase_order_item).get("qty") + qty
-					)
-
-					if total_billed_qty:
-						billed_amt = flt(
-							flt(billed_qty_amt_based_on_po.get(item.purchase_order_item).get("amount"))
-							* (qty / total_billed_qty)
-						)
-					else:
-						billed_amt = 0.0
-
-					# Reduce billed amount based on PO for next iterations
-					billed_qty_amt_based_on_po[item.purchase_order_item]["amount"] -= billed_amt
-
-				if qty:
-					adjusted_amt = flt(billed_amt / qty) * item.qty - flt(item.base_net_amount)
-
-			adjusted_amt = flt(adjusted_amt, item.precision("amount"))
-			pi_landed_cost_amount += adjusted_amt
-			item.db_set("amount_difference_with_purchase_invoice", adjusted_amt, update_modified=False)
-		elif amount and item.billed_amt > amount:
+		if amount and item.billed_amt > amount:
 			per_over_billed = (flt(item.billed_amt / amount, 2) * 100) - 100
 			if (
 				per_over_billed > over_billing_allowance
@@ -278,22 +263,250 @@ def update_billing_percentage(
 					)
 				)
 
-	if pi_landed_cost_amount < 0:
-		total_billed_amount += abs(pi_landed_cost_amount)
-
-	percent_billed = round(100 * (total_billed_amount / (total_amount or 1)), 6)
-	pr_doc.db_set("per_billed", percent_billed)
-
-	if update_modified:
-		pr_doc.set_status(update=True)
-		pr_doc.notify_update()
-
-	if adjust_incoming_rate:
-		adjust_incoming_rate_for_pr(pr_doc)
+	return round(100 * (total_billed_amount / (total_amount or 1)), 6)
 
 
-def get_billed_qty_amount_against_purchase_receipt(pr_doc) -> dict:
-	pr_names = [d.name for d in pr_doc.items]
+def is_billed_by_qty() -> bool:
+	"""Invoice-rate landed cost leaves qty as the only stable measure of billing."""
+	return bool(
+		frappe.db.get_single_value("Buying Settings", "set_landed_cost_based_on_purchase_invoice_rate")
+	)
+
+
+def get_percent_billed_by_qty(
+	pr_doc, items: list, bill_for_rejected: bool, billing_data: frappe._dict
+) -> float:
+	billable_qty = get_billable_qty_by_row(items, billing_data.returned_qty, bill_for_rejected)
+	return get_qty_based_percent_billed(items, billable_qty, get_invoiced_qty(pr_doc, billing_data.invoiced))
+
+
+def get_qty_based_percent_billed(items: list, billable_qty: dict, invoiced_qty: dict) -> float:
+	"""Share of each row's billable qty that is invoiced, weighted by the row's value, or by qty when no row has one."""
+	weigh_by_value = any(flt(item.rate) for item in items)
+
+	total_weight, billed_weight = 0.0, 0.0
+	for item in items:
+		qty = flt(billable_qty.get(item.name))
+		if not qty:
+			continue
+
+		weight = abs(qty * flt(item.rate)) if weigh_by_value else abs(qty)
+		total_weight += weight
+		billed_weight += weight * min(flt(invoiced_qty.get(item.name)) / qty, 1)
+
+	return round(100 * (billed_weight / (total_weight or 1)), 6)
+
+
+def get_billable_qty_by_row(items: list, returned_qty: dict, bill_for_rejected: bool) -> dict:
+	"""Qty left to bill per row; a receipt returned in full is measured against what it received."""
+	billable_qty = {
+		item.name: get_billable_qty(item, returned_qty.get(item.name), bill_for_rejected) for item in items
+	}
+	if any(qty > 0 for qty in billable_qty.values()):
+		return billable_qty
+
+	return {item.name: flt(item.qty) for item in items}
+
+
+def get_billable_qty(item, returned_qty: float | None, bill_for_rejected: bool) -> float:
+	if bill_for_rejected:
+		return flt(item.qty) + flt(item.rejected_qty)
+
+	return flt(item.qty) - flt(returned_qty)
+
+
+def get_invoiced_qty(pr_doc, invoiced: dict) -> dict:
+	invoiced_qty = {name: row.qty for name, row in invoiced.items()}
+
+	for item in pr_doc.items:
+		if item.purchase_invoice_item:
+			invoiced_qty[item.name] = flt(item.qty)
+
+	return invoiced_qty
+
+
+def get_invoiced_qty_and_amount(pr_items: list, bill_for_rejected: bool) -> dict:
+	"""Invoiced qty and base amount per Purchase Receipt Item, direct and through the Purchase Order."""
+	billed = get_billed_qty_amount_against_purchase_receipt([item.name for item in pr_items])
+	invoiced = {
+		pr_detail: frappe._dict(qty=flt(row["qty"]), amount=flt(row["amount"]))
+		for pr_detail, row in billed.items()
+	}
+
+	po_details = list({item.purchase_order_item for item in pr_items if item.purchase_order_item})
+	po_invoice_share = get_po_invoice_share(po_details, bill_for_rejected) if po_details else {}
+
+	for item in pr_items:
+		share = po_invoice_share.get(item.name)
+		if not share or item.purchase_invoice_item:
+			continue
+
+		row = invoiced.setdefault(item.name, frappe._dict(qty=0.0, amount=0.0))
+		row.qty += share.qty
+		row.amount += share.amount
+
+	return invoiced
+
+
+def get_po_invoice_share(po_details: list, bill_for_rejected: bool) -> dict:
+	"""Split invoices made against the Purchase Order over its receipts, oldest invoice to oldest receipt."""
+	po_invoices = get_po_invoices(po_details)
+
+	pr_items = get_purchase_receipts_against_po_details(po_details)
+	pr_item_names = [pr_item.name for pr_item in pr_items]
+	billed_against_pr = get_billed_qty_amount_against_purchase_receipt(pr_item_names)
+	returned_qty = get_item_wise_returned_qty(pr_item_names)
+
+	share = {}
+	for pr_item in pr_items:
+		direct_qty = flt(billed_against_pr.get(pr_item.name, {}).get("qty"))
+		billable_qty = get_billable_qty(pr_item, returned_qty.get(pr_item.name), bill_for_rejected)
+		invoices = po_invoices.get(pr_item.purchase_order_item, [])
+		share[pr_item.name] = take_from_invoices(invoices, billable_qty - direct_qty)
+
+	return share
+
+
+def take_from_invoices(invoices: list, pending_qty: float) -> frappe._dict:
+	"""Take qty from the oldest invoices first, each at its own rate."""
+	taken = frappe._dict(qty=0.0, amount=0.0)
+	for invoice in invoices:
+		qty = min(invoice.qty, pending_qty - taken.qty)
+		if qty <= 0:
+			continue
+
+		amount = invoice.amount * qty / invoice.qty
+		invoice.qty -= qty
+		invoice.amount -= amount
+		taken.qty += qty
+		taken.amount += amount
+
+	return taken
+
+
+def get_po_invoices(po_details: list) -> dict:
+	"""Net qty and base amount of each invoice row made against the Purchase Order, oldest first.
+	Debit note rows net against the row they return; notes that do not update receipt billing are left out."""
+	purchase_invoice = frappe.qb.DocType("Purchase Invoice")
+	purchase_invoice_item = frappe.qb.DocType("Purchase Invoice Item")
+
+	rows = (
+		frappe.qb.from_(purchase_invoice_item)
+		.inner_join(purchase_invoice)
+		.on(purchase_invoice_item.parent == purchase_invoice.name)
+		.select(
+			purchase_invoice_item.name,
+			purchase_invoice_item.po_detail,
+			purchase_invoice_item.qty,
+			purchase_invoice_item.base_net_amount,
+			purchase_invoice_item.purchase_invoice_item,
+			purchase_invoice.is_return,
+		)
+		.where(
+			(purchase_invoice_item.po_detail.isin(po_details))
+			& ((purchase_invoice_item.pr_detail.isnull()) | (purchase_invoice_item.pr_detail == ""))
+			& (purchase_invoice.docstatus == 1)
+			& (purchase_invoice.update_stock == 0)
+			& (
+				(purchase_invoice.is_return == 0)
+				| (purchase_invoice.update_billed_amount_in_purchase_receipt == 1)
+			)
+		)
+		.orderby(CombineDatetime(purchase_invoice.posting_date, purchase_invoice.posting_time))
+		.orderby(purchase_invoice.name)
+		.orderby(purchase_invoice_item.idx)
+	).run(as_dict=True)
+
+	po_invoices = {}
+	for row in rows:
+		invoice_row = row.purchase_invoice_item if row.is_return else row.name
+		invoices = po_invoices.setdefault(row.po_detail, {})
+		invoice = invoices.setdefault(invoice_row, frappe._dict(qty=0.0, amount=0.0))
+		invoice.qty += flt(row.qty)
+		invoice.amount += flt(row.base_net_amount)
+
+	return {
+		po_detail: get_open_invoices(list(invoices.values())) for po_detail, invoices in po_invoices.items()
+	}
+
+
+def get_open_invoices(invoices: list) -> list:
+	"""Debit notes without an invoice row of their own take qty back from the newest rows first, at their own rate."""
+	open_invoices = [invoice for invoice in invoices if invoice.qty > 0]
+	qty_to_take_back = -sum(invoice.qty for invoice in invoices if invoice.qty < 0)
+	amount_to_take_back = -sum(invoice.amount for invoice in invoices if invoice.qty < 0)
+
+	for invoice in reversed(open_invoices):
+		qty = min(invoice.qty, qty_to_take_back)
+		if qty <= 0:
+			break
+
+		amount = amount_to_take_back * qty / qty_to_take_back
+		invoice.qty -= qty
+		invoice.amount -= amount
+		qty_to_take_back -= qty
+		amount_to_take_back -= amount
+
+	return keep_value_of_emptied_invoices(invoices)
+
+
+def keep_value_of_emptied_invoices(invoices: list) -> list:
+	"""A row returned in full at another rate leaves value behind; the newest remaining row keeps it."""
+	remaining = [invoice for invoice in invoices if invoice.qty > 0]
+	leftover = sum(invoice.amount for invoice in invoices if invoice.qty == 0)
+	if remaining and leftover:
+		remaining[-1].amount += leftover
+
+	return remaining
+
+
+def get_invoiced_qty_against_po_items(po_items: list) -> dict:
+	"""Invoiced qty per Purchase Order Item, leaving out returns that do not touch the order."""
+	purchase_invoice = frappe.qb.DocType("Purchase Invoice")
+	purchase_invoice_item = frappe.qb.DocType("Purchase Invoice Item")
+
+	query = (
+		frappe.qb.from_(purchase_invoice_item)
+		.inner_join(purchase_invoice)
+		.on(purchase_invoice_item.parent == purchase_invoice.name)
+		.select(purchase_invoice_item.po_detail, fn.Sum(purchase_invoice_item.qty))
+		.where(
+			(purchase_invoice_item.po_detail.isin(po_items))
+			& (purchase_invoice.docstatus == 1)
+			& (
+				(purchase_invoice.is_return == 0)
+				| (purchase_invoice.update_billed_amount_in_purchase_order == 1)
+			)
+		)
+		.groupby(purchase_invoice_item.po_detail)
+	)
+
+	return frappe._dict(query.run())
+
+
+def set_amount_difference_with_purchase_invoice(items: list, invoiced: dict) -> bool:
+	"""Store each row's gap to its invoiced value; returns True when any row moved."""
+	has_changed = False
+	for item in items:
+		adjusted_amt = 0.0
+		row = invoiced.get(item.name)
+		if row and row.qty:
+			adjusted_amt = flt(row.amount / row.qty) * flt(item.qty) - flt(item.base_net_amount)
+
+		adjusted_amt = flt(adjusted_amt, item.precision("amount"))
+		if adjusted_amt == flt(item.amount_difference_with_purchase_invoice, item.precision("amount")):
+			continue
+
+		item.db_set("amount_difference_with_purchase_invoice", adjusted_amt, update_modified=False)
+		has_changed = True
+
+	return has_changed
+
+
+def get_billed_qty_amount_against_purchase_receipt(pr_names: list) -> dict:
+	if not pr_names:
+		return frappe._dict()
+
 	parent_table = frappe.qb.DocType("Purchase Invoice")
 	table = frappe.qb.DocType("Purchase Invoice Item")
 	query = (
@@ -305,7 +518,11 @@ def get_billed_qty_amount_against_purchase_receipt(pr_doc) -> dict:
 			fn.Sum(table.base_net_amount).as_("amount"),
 			fn.Sum(table.qty).as_("qty"),
 		)
-		.where((table.pr_detail.isin(pr_names)) & (table.docstatus == 1))
+		.where(
+			(table.pr_detail.isin(pr_names))
+			& (table.docstatus == 1)
+			& ((parent_table.is_return == 0) | (parent_table.update_billed_amount_in_purchase_receipt == 1))
+		)
 		.groupby(table.pr_detail)
 	)
 	invoice_data = query.run(as_dict=1)
@@ -325,49 +542,6 @@ def get_billed_qty_amount_against_purchase_receipt(pr_doc) -> dict:
 	return billed_qty_amt
 
 
-def get_billed_qty_amount_against_purchase_order(pr_doc) -> dict:
-	po_names = list(
-		set(
-			[
-				d.purchase_order_item
-				for d in pr_doc.items
-				if d.purchase_order_item and not d.purchase_invoice_item
-			]
-		)
-	)
-
-	invoice_data_po_based = frappe._dict()
-	if po_names:
-		parent_table = frappe.qb.DocType("Purchase Invoice")
-		table = frappe.qb.DocType("Purchase Invoice Item")
-
-		query = (
-			frappe.qb.from_(parent_table)
-			.inner_join(table)
-			.on(parent_table.name == table.parent)
-			.select(
-				table.po_detail,
-				fn.Sum(table.qty).as_("qty"),
-				fn.Sum(table.base_net_amount).as_("amount"),
-			)
-			.where((table.po_detail.isin(po_names)) & (table.docstatus == 1) & (table.pr_detail.isnull()))
-			.groupby(table.po_detail)
-		)
-
-		invoice_data = query.run(as_dict=1)
-		if not invoice_data:
-			return frappe._dict()
-
-		for row in invoice_data:
-			if row.po_detail not in invoice_data_po_based:
-				invoice_data_po_based[row.po_detail] = {"amount": 0, "qty": 0}
-
-			invoice_data_po_based[row.po_detail]["amount"] += flt(row.amount)
-			invoice_data_po_based[row.po_detail]["qty"] += flt(row.qty)
-
-	return invoice_data_po_based
-
-
 def adjust_incoming_rate_for_pr(doc) -> None:
 	doc.update_valuation_rate(reset_outgoing_rate=False)
 
@@ -380,9 +554,7 @@ def adjust_incoming_rate_for_pr(doc) -> None:
 	doc.repost_future_sle_and_gle(force=True)
 
 
-def get_item_wise_returned_qty(pr_doc) -> dict:
-	items = [d.name for d in pr_doc.items]
-
+def get_item_wise_returned_qty(items: list) -> dict:
 	return frappe._dict(
 		frappe.get_all(
 			"Purchase Receipt",

erpnext/stock/doctype/purchase_receipt/test_purchase_receipt.py

@@ -1117,6 +1117,271 @@ class TestPurchaseReceipt(ERPNextTestSuite):
 		po.reload()
 		po.cancel()
 
+	@ERPNextTestSuite.change_settings(
+		"Buying Settings", {"maintain_same_rate": 0, "set_landed_cost_based_on_purchase_invoice_rate": 1}
+	)
+	def test_per_billed_by_qty_when_landed_cost_follows_invoice_rate(self):
+		pr = make_purchase_receipt(qty=100, rate=50)
+		pi = make_purchase_invoice(pr.name)
+		pi.items[0].qty = 25
+		pi.items[0].rate = 200
+		pi.submit()
+
+		pr.reload()
+		self.assertEqual(pr.per_billed, 25)
+		self.assertEqual(pr.status, "Partly Billed")
+
+	@ERPNextTestSuite.change_settings(
+		"Buying Settings", {"maintain_same_rate": 0, "set_landed_cost_based_on_purchase_invoice_rate": 1}
+	)
+	def test_po_invoice_qty_spread_fifo_when_landed_cost_follows_invoice_rate(self):
+		from erpnext.buying.doctype.purchase_order.test_purchase_order import create_purchase_order
+
+		po = create_purchase_order(qty=100, rate=50)
+		receipts = make_receipts_against_order(po.name, ((60, "08:00"), (40, "10:00")))
+		make_invoice_against_order(po.name, qty=70, rate=40)
+
+		for pr in receipts:
+			pr.reload()
+
+		self.assertEqual(receipts[0].per_billed, 100)
+		self.assertEqual(receipts[1].per_billed, 25)
+
+	@ERPNextTestSuite.change_settings(
+		"Buying Settings",
+		{
+			"maintain_same_rate": 0,
+			"set_landed_cost_based_on_purchase_invoice_rate": 1,
+			"bill_for_rejected_quantity_in_purchase_invoice": 0,
+		},
+	)
+	def test_fully_returned_receipt_skipped_in_po_invoice_split(self):
+		from erpnext.buying.doctype.purchase_order.test_purchase_order import create_purchase_order
+		from erpnext.stock.doctype.purchase_receipt.mapper import make_purchase_return
+
+		po = create_purchase_order(qty=100, rate=50)
+		receipts = make_receipts_against_order(po.name, ((60, "08:00"), (40, "10:00")))
+		make_purchase_return(receipts[0].name).submit()
+		make_invoice_against_order(po.name, qty=40, rate=50)
+
+		for pr in receipts:
+			pr.reload()
+
+		self.assertEqual(receipts[0].per_billed, 0)
+		self.assertEqual(receipts[1].per_billed, 100)
+
+	@ERPNextTestSuite.change_settings(
+		"Buying Settings",
+		{
+			"maintain_same_rate": 0,
+			"set_landed_cost_based_on_purchase_invoice_rate": 1,
+			"bill_for_rejected_quantity_in_purchase_invoice": 0,
+		},
+	)
+	def test_fully_returned_row_left_out_of_qty_billing(self):
+		from erpnext.stock.doctype.purchase_receipt.mapper import make_purchase_return
+
+		pr = make_purchase_receipt(qty=10, rate=50, do_not_save=True)
+		pr.append("items", pr.items[0].as_dict(no_default_fields=True))
+		pr.submit()
+		returned_row, invoiced_row = pr.items
+
+		pr_return = make_purchase_return(pr.name)
+		pr_return.set(
+			"items", [row for row in pr_return.items if row.purchase_receipt_item == returned_row.name]
+		)
+		pr_return.submit()
+
+		pi = make_purchase_invoice(pr.name)
+		pi.set("items", [row for row in pi.items if row.pr_detail == invoiced_row.name])
+		pi.submit()
+
+		pr.reload()
+		self.assertEqual(pr.per_billed, 100)
+
+	@ERPNextTestSuite.change_settings(
+		"Buying Settings", {"maintain_same_rate": 0, "set_landed_cost_based_on_purchase_invoice_rate": 1}
+	)
+	def test_zero_rate_receipt_billed_by_qty(self):
+		pr = make_purchase_receipt(item_code="_Test Non Stock Item", qty=10, rate=0)
+		pi = make_purchase_invoice(pr.name)
+		pi.items[0].rate = 5
+		pi.submit()
+
+		pr.reload()
+		self.assertEqual(pr.per_billed, 100)
+
+	@ERPNextTestSuite.change_settings(
+		"Buying Settings", {"maintain_same_rate": 0, "set_landed_cost_based_on_purchase_invoice_rate": 1}
+	)
+	def test_non_updating_debit_note_kept_out_of_qty_billing(self):
+		from erpnext.accounts.doctype.purchase_invoice.mapper import make_debit_note
+
+		pr = make_purchase_receipt(qty=100, rate=50)
+		pi = make_purchase_invoice(pr.name)
+		pi.items[0].qty = 50
+		pi.submit()
+
+		debit_note = make_debit_note(pi.name)
+		debit_note.items[0].qty = -20
+		debit_note.update_billed_amount_in_purchase_receipt = 0
+		debit_note.submit()
+
+		pi = make_purchase_invoice(pr.name)
+		pi.items[0].qty = 50
+		pi.submit()
+
+		pr.reload()
+		self.assertEqual(pr.per_billed, 100)
+
+	@ERPNextTestSuite.change_settings(
+		"Buying Settings", {"maintain_same_rate": 0, "set_landed_cost_based_on_purchase_invoice_rate": 1}
+	)
+	def test_landed_cost_takes_order_invoices_oldest_first(self):
+		from erpnext.buying.doctype.purchase_order.test_purchase_order import create_purchase_order
+
+		po = create_purchase_order(qty=100, rate=50)
+		receipts = make_receipts_against_order(po.name, ((60, "08:00"), (40, "10:00")))
+
+		make_invoice_against_order(po.name, qty=60, rate=70)
+		self.assertEqual(get_amount_differences(receipts), [1200, 0])
+
+		make_invoice_against_order(po.name, qty=40, rate=55)
+		self.assertEqual(get_amount_differences(receipts), [1200, 200])
+
+	@ERPNextTestSuite.change_settings(
+		"Buying Settings", {"maintain_same_rate": 0, "set_landed_cost_based_on_purchase_invoice_rate": 1}
+	)
+	def test_landed_cost_takes_invoice_rows_in_order(self):
+		from erpnext.buying.doctype.purchase_order.mapper import (
+			make_purchase_invoice as make_purchase_invoice_from_po,
+		)
+		from erpnext.buying.doctype.purchase_order.test_purchase_order import create_purchase_order
+
+		po = create_purchase_order(qty=100, rate=50)
+		receipts = make_receipts_against_order(po.name, ((60, "08:00"), (40, "10:00")))
+
+		pi = make_purchase_invoice_from_po(po.name)
+		pi.append("items", pi.items[0].as_dict(no_default_fields=True))
+		pi.items[0].qty = 60
+		pi.items[0].rate = 70
+		pi.items[1].qty = 40
+		pi.items[1].rate = 55
+		pi.submit()
+
+		self.assertEqual(get_amount_differences(receipts), [1200, 200])
+
+	@ERPNextTestSuite.change_settings(
+		"Buying Settings", {"maintain_same_rate": 0, "set_landed_cost_based_on_purchase_invoice_rate": 1}
+	)
+	def test_non_updating_order_debit_note_kept_out_of_invoice_split(self):
+		from erpnext.accounts.doctype.purchase_invoice.mapper import make_debit_note
+		from erpnext.buying.doctype.purchase_order.mapper import (
+			make_purchase_invoice as make_purchase_invoice_from_po,
+		)
+		from erpnext.buying.doctype.purchase_order.test_purchase_order import create_purchase_order
+		from erpnext.stock.doctype.purchase_receipt.services.billing_status import update_billing_percentage
+
+		po = create_purchase_order(qty=100, rate=50)
+		(pr,) = make_receipts_against_order(po.name, ((100, "08:00"),))
+		pi = make_purchase_invoice_from_po(po.name)
+		pi.submit()
+
+		debit_note = make_debit_note(pi.name)
+		debit_note.items[0].qty = -20
+		debit_note.update_billed_amount_in_purchase_receipt = 0
+		debit_note.submit()
+
+		update_billing_percentage(frappe.get_doc("Purchase Receipt", pr.name))
+		pr.reload()
+		self.assertEqual(pr.per_billed, 100)
+
+	@ERPNextTestSuite.change_settings(
+		"Buying Settings", {"maintain_same_rate": 0, "set_landed_cost_based_on_purchase_invoice_rate": 1}
+	)
+	def test_unlinked_order_debit_note_takes_back_newest_invoice_qty(self):
+		from erpnext.buying.doctype.purchase_order.mapper import (
+			make_purchase_invoice as make_purchase_invoice_from_po,
+		)
+		from erpnext.buying.doctype.purchase_order.test_purchase_order import create_purchase_order
+
+		po = create_purchase_order(qty=100, rate=50)
+		receipts = make_receipts_against_order(po.name, ((60, "08:00"), (40, "10:00")))
+		pi = make_purchase_invoice_from_po(po.name)
+		pi.submit()
+
+		debit_note = frappe.copy_doc(pi)
+		debit_note.is_return = 1
+		debit_note.items[0].qty = -40
+		debit_note.submit()
+
+		for pr in receipts:
+			pr.reload()
+
+		self.assertEqual([pr.per_billed for pr in receipts], [100, 0])
+
+	@ERPNextTestSuite.change_settings(
+		"Buying Settings", {"maintain_same_rate": 0, "set_landed_cost_based_on_purchase_invoice_rate": 1}
+	)
+	def test_unlinked_order_debit_note_takes_back_its_own_value(self):
+		from erpnext.buying.doctype.purchase_order.mapper import (
+			make_purchase_invoice as make_purchase_invoice_from_po,
+		)
+		from erpnext.buying.doctype.purchase_order.test_purchase_order import create_purchase_order
+
+		po = create_purchase_order(qty=100, rate=50)
+		receipts = make_receipts_against_order(po.name, ((100, "08:00"),))
+		pi = make_purchase_invoice_from_po(po.name)
+		pi.submit()
+
+		debit_note = frappe.copy_doc(pi)
+		debit_note.is_return = 1
+		debit_note.items[0].qty = -20
+		debit_note.items[0].rate = 60
+		debit_note.submit()
+
+		self.assertEqual(get_amount_differences(receipts), [-250])
+
+	@ERPNextTestSuite.change_settings(
+		"Buying Settings", {"maintain_same_rate": 0, "set_landed_cost_based_on_purchase_invoice_rate": 1}
+	)
+	def test_landed_cost_refreshed_on_receipt_whose_share_moved(self):
+		from erpnext.buying.doctype.purchase_order.test_purchase_order import create_purchase_order
+
+		po = create_purchase_order(qty=100, rate=50)
+		receipts = make_receipts_against_order(po.name, ((60, "08:00"), (40, "10:00")))
+
+		pi = make_purchase_invoice(receipts[0].name)
+		pi.items[0].qty = 1
+		pi.submit()
+		make_invoice_against_order(po.name, qty=69, rate=40)
+
+		second_row = receipts[1].items[0].name
+		self.assertEqual(frappe.db.get_value("Purchase Receipt Item", second_row, "billed_amt"), 0)
+		self.assertEqual(
+			frappe.db.get_value(
+				"Purchase Receipt Item", second_row, "amount_difference_with_purchase_invoice"
+			),
+			-400,
+		)
+
+	@ERPNextTestSuite.change_settings(
+		"Buying Settings", {"maintain_same_rate": 0, "set_landed_cost_based_on_purchase_invoice_rate": 1}
+	)
+	def test_invoice_refresh_leaves_unchanged_receipt_alone(self):
+		from erpnext.buying.doctype.purchase_order.test_purchase_order import create_purchase_order
+
+		po = create_purchase_order(qty=100, rate=50)
+		receipts = make_receipts_against_order(po.name, ((60, "08:00"), (40, "10:00")))
+		untouched_modified = frappe.db.get_value("Purchase Receipt", receipts[1].name, "modified")
+
+		make_invoice_against_order(po.name, qty=60, rate=50)
+
+		self.assertEqual(frappe.db.get_value("Purchase Receipt", receipts[0].name, "per_billed"), 100)
+		self.assertEqual(
+			frappe.db.get_value("Purchase Receipt", receipts[1].name, "modified"), untouched_modified
+		)
+
 	def test_serial_no_against_purchase_receipt(self):
 		item_code = "Test Manual Created Serial No"
 		if not frappe.db.exists("Item", item_code):
@@ -7771,6 +8036,40 @@ def get_items(**args):
 	]
 
 
+def make_receipts_against_order(purchase_order: str, receipts: tuple) -> list:
+	from erpnext.buying.doctype.purchase_order.mapper import make_purchase_receipt as make_receipt_from_order
+
+	receipt_docs = []
+	for qty, posting_time in receipts:
+		pr = make_receipt_from_order(purchase_order)
+		pr.set_posting_time = 1
+		pr.posting_time = posting_time
+		pr.items[0].received_qty = qty
+		pr.items[0].qty = qty
+		pr.submit()
+		receipt_docs.append(pr)
+
+	return receipt_docs
+
+
+def get_amount_differences(receipts: list) -> list:
+	return [
+		frappe.db.get_value(
+			"Purchase Receipt Item", pr.items[0].name, "amount_difference_with_purchase_invoice"
+		)
+		for pr in receipts
+	]
+
+
+def make_invoice_against_order(purchase_order: str, qty: float, rate: float) -> None:
+	from erpnext.buying.doctype.purchase_order.mapper import make_purchase_invoice as make_invoice_from_order
+
+	pi = make_invoice_from_order(purchase_order)
+	pi.items[0].qty = qty
+	pi.items[0].rate = rate
+	pi.submit()
+
+
 def make_purchase_receipt(**args):
 	frappe.db.set_single_value("Buying Settings", "allow_multiple_items", 1)
 	pr = frappe.new_doc("Purchase Receipt")

erpnext/stock/doctype/stock_entry/services/manufacturing.py

@@ -126,37 +126,15 @@ class BaseManufactureStockEntry(BaseStockEntry):
 			self.doc.append("items", item_args)
 
 	def set_process_loss_qty(self):
-		precision = self.doc.precision("process_loss_qty")
-		if self.doc.work_order:
-			data = frappe.get_all(
-				"Work Order Operation",
-				filters={"parent": self.doc.work_order},
-				fields=[{"MAX": "process_loss_qty", "as": "process_loss_qty"}],
-			)
-
-			if data and data[0].process_loss_qty:
-				process_loss_qty = data[0].process_loss_qty
-				if flt(self.doc.process_loss_qty, precision) != flt(process_loss_qty, precision):
-					self.doc.process_loss_qty = flt(process_loss_qty, precision)
-
-					frappe.msgprint(
-						_("The Process Loss Qty has been reset as per the Job Card's Process Loss Qty"),
-						alert=True,
-					)
-
-		if not self.doc.process_loss_percentage and not self.doc.process_loss_qty:
-			self.doc.process_loss_percentage = frappe.get_cached_value(
+		process_loss_qty = self.doc.get_pending_process_loss_qty()
+		if not process_loss_qty:
+			process_loss_percentage = frappe.get_cached_value(
 				"BOM", self.doc.bom_no, "process_loss_percentage"
 			)
+			process_loss_qty = flt(self.doc.fg_completed_qty) * flt(process_loss_percentage) / 100
 
-		if self.doc.process_loss_percentage and not self.doc.process_loss_qty:
-			self.doc.process_loss_qty = flt(
-				(flt(self.doc.fg_completed_qty) * flt(self.doc.process_loss_percentage)) / 100
-			)
-		elif self.doc.process_loss_qty and self.doc.fg_completed_qty:
-			self.doc.process_loss_percentage = flt(
-				(flt(self.doc.process_loss_qty) / flt(self.doc.fg_completed_qty)) * 100
-			)
+		self.doc.process_loss_qty = flt(process_loss_qty, self.doc.precision("process_loss_qty"))
+		self.doc.set_process_loss_percentage()
 
 	def add_finished_goods(self):
 		item_details = get_production_item_details(self.doc.work_order, self.doc.bom_no)

erpnext/stock/doctype/stock_entry/stock_entry.js

@@ -965,26 +965,6 @@ frappe.ui.form.on("Stock Entry", {
 		}
 	},
 
-	process_loss_qty(frm) {
-		if (frm.doc.process_loss_qty) {
-			frm.doc.process_loss_percentage = flt(
-				(frm.doc.process_loss_qty / frm.doc.fg_completed_qty) * 100,
-				precision("process_loss_qty", frm.doc)
-			);
-			refresh_field("process_loss_percentage");
-		}
-	},
-
-	process_loss_percentage(frm) {
-		if (frm.doc.process_loss_percentage) {
-			frm.doc.process_loss_qty = flt(
-				(frm.doc.fg_completed_qty * frm.doc.process_loss_percentage) / 100,
-				precision("process_loss_qty", frm.doc)
-			);
-			refresh_field("process_loss_qty");
-		}
-	},
-
 	set_fg_completed_qty(frm) {
 		let fg_completed_qty = 0;
 

erpnext/stock/doctype/stock_entry/stock_entry.json

@@ -35,10 +35,8 @@
   "cb1",
   "fg_completed_qty",
   "get_items",
-  "section_break_7qsm",
-  "process_loss_percentage",
-  "column_break_e92r",
   "process_loss_qty",
+  "process_loss_percentage",
   "section_break_jwgn",
   "from_warehouse",
   "source_warehouse_address",
@@ -694,28 +692,19 @@
    "label": "Details",
    "oldfieldtype": "Section Break"
   },
-  {
-   "collapsible": 1,
-   "depends_on": "eval: doc.fg_completed_qty > 0 && in_list([\"Manufacture\", \"Repack\"], doc.purpose)",
-   "fieldname": "section_break_7qsm",
-   "fieldtype": "Section Break",
-   "label": "Process Loss"
-  },
   {
    "depends_on": "eval: doc.fg_completed_qty > 0 && in_list([\"Manufacture\", \"Repack\"], doc.purpose)",
    "fieldname": "process_loss_qty",
    "fieldtype": "Float",
-   "label": "Process Loss Qty"
+   "label": "Process Loss Qty",
+   "read_only": 1
   },
   {
-   "fieldname": "column_break_e92r",
-   "fieldtype": "Column Break"
-  },
-  {
-   "depends_on": "eval:doc.from_bom && doc.fg_completed_qty",
+   "depends_on": "eval:doc.from_bom && doc.fg_completed_qty > 0 && in_list([\"Manufacture\", \"Repack\"], doc.purpose)",
    "fieldname": "process_loss_percentage",
    "fieldtype": "Percent",
-   "label": "% Process Loss"
+   "label": "% Process Loss",
+   "read_only": 1
   },
   {
    "fieldname": "items_section",
@@ -796,7 +785,7 @@
  "index_web_pages_for_search": 1,
  "is_submittable": 1,
  "links": [],
- "modified": "2026-08-28 18:00:00.000000",
+ "modified": "2026-09-26 16:44:19.045190",
  "modified_by": "Administrator",
  "module": "Stock",
  "name": "Stock Entry",

erpnext/stock/doctype/stock_entry/stock_entry.py

@@ -3,7 +3,6 @@
 
 
 import json
-from collections import defaultdict
 
 import frappe
 from frappe import _
@@ -316,7 +315,6 @@ class StockEntry(StockController, SubcontractingInwardController):
 		sbb.validate_warehouse_of_sabb()
 		self.validate_source_stock_entry()
 		self.validate_bom()
-		self.set_process_loss_qty()
 		self.validate_company_in_accounting_dimension()
 
 		if self.purpose in ("Manufacture", "Repack"):
@@ -342,7 +340,7 @@ class StockEntry(StockController, SubcontractingInwardController):
 
 		self.validate_batch()
 		self.validate_inspection()
-		self.validate_fg_completed_qty()
+		self.set_process_loss_qty()
 		self.validate_job_card_pending_production()
 		self.validate_difference_account()
 		self.validate_job_card_item()
@@ -517,41 +515,6 @@ class StockEntry(StockController, SubcontractingInwardController):
 
 			item.validate_and_update_item_details(item_details, self.company, self.purpose)
 
-	def validate_fg_completed_qty(self):
-		if self.purpose != "Manufacture" or not self.from_bom:
-			return
-		fg_qty = self._aggregate_fg_qty()
-		if fg_qty:
-			self._check_process_loss_qty(fg_qty)
-
-	def _aggregate_fg_qty(self):
-		fg_qty = defaultdict(float)
-		for d in self.items:
-			if d.is_finished_item:
-				fg_qty[d.item_code] += flt(d.qty)
-		return fg_qty
-
-	def _check_process_loss_qty(self, fg_qty):
-		precision = frappe.get_precision("Stock Entry Detail", "qty")
-		fg_item = next(iter(fg_qty.keys()))
-		fg_item_qty = flt(fg_qty[fg_item], precision)
-		fg_completed_qty = flt(self.fg_completed_qty, precision)
-		for d in self.items:
-			if fg_qty.get(d.item_code):
-				self._validate_fg_qty_with_process_loss(d, fg_item_qty, fg_completed_qty, precision)
-
-	def _validate_fg_qty_with_process_loss(self, d, fg_item_qty, fg_completed_qty, precision):
-		if (fg_completed_qty - fg_item_qty) > 0:
-			self.process_loss_qty = fg_completed_qty - fg_item_qty
-		if not self.process_loss_qty:
-			return
-		if fg_completed_qty != (flt(fg_item_qty, precision) + flt(self.process_loss_qty, precision)):
-			frappe.throw(
-				_(
-					"Since there is a process loss of {0} units for the finished good {1}, you should reduce the quantity by {0} units for the finished good {1} in the Items Table."
-				).format(frappe.bold(self.process_loss_qty), frappe.bold(d.item_code))
-			)
-
 	def validate_difference_account(self):
 		if not cint(erpnext.is_perpetual_inventory_enabled(self.company)):
 			return
@@ -1567,6 +1530,78 @@ class StockEntry(StockController, SubcontractingInwardController):
 		if self.purpose not in ("Manufacture", "Repack"):
 			return
 
+		if self.from_bom and self.bom_no and flt(self.fg_completed_qty) and not self.is_fg_conversion:
+			self.set_process_loss_from_finished_goods()
+		else:
+			self.reset_process_loss_to_pending_qty()
+
+	def set_process_loss_from_finished_goods(self):
+		"""Loss is the part of Finished Good Quantity the BOM item rows do not cover."""
+		process_loss_qty = max(flt(self.fg_completed_qty) - self.get_bom_item_finished_qty(), 0)
+		self.process_loss_qty = flt(process_loss_qty, self.precision("process_loss_qty"))
+		self.set_process_loss_percentage()
+
+	def get_bom_item_finished_qty(self):
+		"""Received qty of the BOM item and its variants. Other Repack outputs do not count."""
+		bom_item = frappe.get_cached_value("BOM", self.bom_no, "item")
+		received_rows = [
+			row for row in self.items if row.is_finished_item and row.t_warehouse and not row.s_warehouse
+		]
+		bom_outputs = [bom_item, *self.get_variants_of(bom_item, {row.item_code for row in received_rows})]
+		bom_item_rows = [row for row in received_rows if row.item_code in bom_outputs]
+
+		if not bom_item_rows:
+			frappe.throw(
+				_(
+					"This entry is made from BOM {0}, so its finished good must be {1} or a variant of it. Uncheck From BOM to make another item."
+				).format(frappe.bold(self.bom_no), frappe.bold(bom_item)),
+				title=_("Finished Good Does Not Match BOM"),
+				exc=FinishedGoodError,
+			)
+
+		if not self.work_order:
+			self.validate_finished_good_stock_uom(bom_item, bom_item_rows)
+
+		return sum(flt(row.transfer_qty) for row in bom_item_rows)
+
+	def validate_finished_good_stock_uom(self, bom_item, rows):
+		"""Without a work order, Finished Good Quantity is in the BOM item's stock UOM."""
+		stock_uom = frappe.get_cached_value("Item", bom_item, "stock_uom")
+		for row in rows:
+			if row.stock_uom != stock_uom:
+				frappe.throw(
+					_(
+						"Row {0}: {1} has stock UOM {2}, but Finished Good Quantity is in {3}, the stock UOM of {4}. Make it through a Work Order, or uncheck From BOM."
+					).format(
+						row.idx,
+						frappe.bold(row.item_code),
+						frappe.bold(row.stock_uom),
+						frappe.bold(stock_uom),
+						frappe.bold(bom_item),
+					),
+					title=_("Finished Good UOM Mismatch"),
+					exc=FinishedGoodError,
+				)
+
+	def get_variants_of(self, template, item_codes):
+		other_items = item_codes - {template}
+		if not other_items:
+			return []
+
+		return frappe.get_all(
+			"Item", filters={"name": ["in", list(other_items)], "variant_of": template}, pluck="name"
+		)
+
+	def set_process_loss_percentage(self):
+		if not flt(self.fg_completed_qty):
+			return
+
+		self.process_loss_percentage = flt(
+			flt(self.process_loss_qty) / flt(self.fg_completed_qty) * 100,
+			self.precision("process_loss_percentage"),
+		)
+
+	def reset_process_loss_to_pending_qty(self):
 		precision = self.precision("process_loss_qty")
 		process_loss_qty = self.get_pending_process_loss_qty()
 		if process_loss_qty and flt(self.process_loss_qty, precision) != flt(process_loss_qty, precision):
@@ -1577,20 +1612,6 @@ class StockEntry(StockController, SubcontractingInwardController):
 				alert=True,
 			)
 
-		if not self.process_loss_percentage and not self.process_loss_qty:
-			self.process_loss_percentage = frappe.get_cached_value(
-				"BOM", self.bom_no, "process_loss_percentage"
-			)
-
-		if self.process_loss_percentage and not self.process_loss_qty:
-			self.process_loss_qty = flt(
-				(flt(self.fg_completed_qty) * flt(self.process_loss_percentage)) / 100
-			)
-		elif self.process_loss_qty and self.fg_completed_qty:
-			self.process_loss_percentage = flt(
-				(flt(self.process_loss_qty) / flt(self.fg_completed_qty)) * 100
-			)
-
 	def validate_job_card_pending_production(self):
 		"""A draft created before other entries were submitted must not book more than the job
 		card still has left; without this, a stale draft over-produces the finished good."""

erpnext/stock/doctype/stock_entry/test_stock_entry.py

@@ -4881,27 +4881,149 @@ class TestStockEntryCoverage(ERPNextTestSuite):
 		frappe.db.set_value("Work Order", wo.name, "produced_qty", wo.qty)
 		self.assertNotIn(wo.name, pending_work_orders())
 
-	def test_process_loss_percentage_resyncs_from_qty(self):
-		# changing fg qty recomputes process_loss_qty and process_loss_percentage
-		se = frappe.new_doc("Stock Entry")
-		se.purpose = "Manufacture"
+	def make_process_loss_entry(self, purpose="Manufacture", fg_item=None):
+		"""Entry for 100 units from a BOM with 5% process loss, items fetched but not saved."""
+		from erpnext.manufacturing.doctype.production_plan.test_production_plan import make_bom
+
+		fg_item = fg_item or make_item("Process Loss FG", properties={"is_stock_item": 1}).name
+		rm_item = make_item("Process Loss RM", properties={"is_stock_item": 1}).name
+
+		se = make_stock_entry(
+			item_code=fg_item, qty=100, purpose=purpose, company="_Test Company", do_not_save=True
+		)
+		se.items = []
+		se.from_bom = 1
+		se.bom_no = make_bom(
+			item=fg_item, quantity=100, raw_materials=[rm_item], process_loss_percentage=5
+		).name
+		se.fg_completed_qty = 100
+		se.from_warehouse = "_Test Warehouse - _TC"
+		se.to_warehouse = "_Test Warehouse 1 - _TC"
+		se.get_items()
+		return se
+
+	def get_finished_good_row(self, se):
+		return next(row for row in se.items if row.is_finished_item)
+
+	def test_process_loss_follows_finished_good_qty(self):
+		for purpose in ("Manufacture", "Repack"):
+			se = self.make_process_loss_entry(purpose)
+			self.assertEqual(self.get_finished_good_row(se).qty, 95)
+
+			self.get_finished_good_row(se).qty = 90
+			se.save()
+
+			self.assertEqual(se.process_loss_qty, 10)
+			self.assertEqual(se.process_loss_percentage, 10)
+
+	def test_process_loss_counts_variant_of_bom_item(self):
+		make_item_variant()
+		se = self.make_process_loss_entry(fg_item="_Test Variant Item")
+		self.get_finished_good_row(se).item_code = "_Test Variant Item-S"
+		self.get_finished_good_row(se).qty = 90
+		se.save()
+
+		self.assertEqual(se.process_loss_qty, 10)
+
+	def test_from_bom_entry_rejects_variant_in_another_stock_uom(self):
+		from erpnext.controllers.item_variant import create_variant
+
+		frappe.delete_doc_if_exists("Item", "_Test Variant Item-L", force=1)
+		with self.change_settings("Item Variant Settings", {"allow_different_uom": 1}):
+			variant = create_variant("_Test Variant Item", {"Test Size": "Large"})
+			variant.stock_uom = "_Test UOM 1"
+			variant.insert()
+
+		se = self.make_process_loss_entry(fg_item="_Test Variant Item")
+		fg_row = self.get_finished_good_row(se)
+		fg_row.item_code = variant.name
+		fg_row.uom = "_Test UOM 1"
+		fg_row.conversion_factor = 1
+
+		self.assertRaisesRegex(FinishedGoodError, "has stock UOM", se.save)
+
+	def test_from_bom_entry_rejects_alternative_of_bom_item(self):
+		properties = {"is_stock_item": 1, "allow_alternative_item": 1}
+		fg_item = make_item("Process Loss Alternative Source", properties=properties).name
+		alternative = make_item("Process Loss Alternative FG", properties=properties).name
+		frappe.get_doc(
+			{"doctype": "Item Alternative", "item_code": fg_item, "alternative_item_code": alternative}
+		).insert()
+
+		se = self.make_process_loss_entry(fg_item=fg_item)
+		self.get_finished_good_row(se).item_code = alternative
+
+		self.assertRaises(FinishedGoodError, se.save)
+
+	def test_process_loss_ignores_finished_flag_on_source_row(self):
+		se = self.make_process_loss_entry()
+		se.append(
+			"items",
+			{
+				"item_code": self.get_finished_good_row(se).item_code,
+				"qty": 5,
+				"uom": "Nos",
+				"conversion_factor": 1,
+				"s_warehouse": "_Test Warehouse - _TC",
+				"is_finished_item": 1,
+			},
+		)
+		se.to_warehouse = None
+		se.save()
+
+		self.assertEqual(se.process_loss_qty, 5)
+
+	def test_from_bom_entry_rejects_finished_item_other_than_bom_item(self):
+		other_item = make_item("Process Loss Unrelated FG", properties={"is_stock_item": 1}).name
+		for purpose in ("Manufacture", "Repack"):
+			se = self.make_process_loss_entry(purpose)
+			self.get_finished_good_row(se).item_code = other_item
+
+			self.assertRaises(FinishedGoodError, se.save)
+
+	def test_process_loss_ignores_other_repack_outputs(self):
+		other_item = make_item("Process Loss Other Output", properties={"is_stock_item": 1}).name
+		se = self.make_process_loss_entry("Repack")
+		fg_row = self.get_finished_good_row(se)
+		fg_row.set_basic_rate_manually = 1
+		fg_row.basic_rate = 10
+		other_output = se.append(
+			"items",
+			{
+				"item_code": other_item,
+				"qty": 10,
+				"uom": "Nos",
+				"conversion_factor": 1,
+				"t_warehouse": "_Test Warehouse 1 - _TC",
+				"is_finished_item": 1,
+				"set_basic_rate_manually": 1,
+				"basic_rate": 10,
+			},
+		)
+		se.items.remove(other_output)
+		se.items.insert(0, other_output)
+		se.save()
+
+		self.assertEqual(se.process_loss_qty, 5)
+
+	def test_zero_process_loss_saves_despite_bom_percentage(self):
+		se = self.make_process_loss_entry()
+		self.get_finished_good_row(se).qty = 100
+		se.save()
+
+		self.assertEqual(se.process_loss_qty, 0)
+		self.assertEqual(se.process_loss_percentage, 0)
+
+	def test_get_items_recomputes_process_loss_from_bom(self):
+		se = self.make_process_loss_entry()
+		se.save()
+
 		se.fg_completed_qty = 200
-		se.process_loss_qty = 100
-		se.process_loss_percentage = 80
+		se.get_items()
 
-		se.set_process_loss_qty()
-
-		self.assertEqual(se.process_loss_percentage, 50)
-
-	def test_process_loss_qty_derived_from_percentage_when_qty_blank(self):
-		se = frappe.new_doc("Stock Entry")
-		se.purpose = "Manufacture"
-		se.fg_completed_qty = 200
-		se.process_loss_percentage = 25
-
-		se.set_process_loss_qty()
-
-		self.assertEqual(se.process_loss_qty, 50)
+		self.assertEqual(se.process_loss_qty, 10)
+		self.assertEqual(se.process_loss_percentage, 5)
+		self.assertEqual(self.get_finished_good_row(se).qty, 190)
 
 
 def make_serialized_item(self, **args):