test(manufacturing): cover per-attribution returns and consumption drain

add_materials_from_transfer() merged transfer rows of the same item and
warehouse into one bucket even when their original_item differed, keeping
only the first row's attribution. Return entries and Manufacture backflush
rows built from a merged bucket then charged the whole quantity to a single
requirement, so per-row returned_qty and net transfer coverage landed on
the wrong required item.

Buckets are now keyed by (item_code, warehouse, original_item). Manufacture
consumption rows carry no original_item, so consumption is distributed
across the item's buckets in transfer order; any residual lands on the last
bucket, preserving the previous net quantity. Serial and batch deductions
resolve the bucket that actually holds the serial no or batch.

Restore the pre-split-refactor guard that skipped zero-qty rows when
building entries from buckets (dropped in e4b5e6bd1e); without it a fully
consumed bucket would emit a zero-qty row and block submission.

erpnext/manufacturing/doctype/work_order/services/required_items.py

@@ -23,6 +23,8 @@ from erpnext.manufacturing.doctype.work_order.services.reservation import (
 from erpnext.manufacturing.doctype.work_order.services.status import StatusService
 from erpnext.stock.utils import get_bin, get_latest_stock_qty
 
+_FULL_TRANSFER_TOLERANCE = 0.001
+
 
 class RequiredItemsService:
 	def __init__(self, doc):
@@ -148,7 +150,7 @@ class RequiredItemsService:
 					row, transferred_qty, row_wise_serial_batch
 				)
 
-		self.recompute_material_transferred_for_manufacturing(transferred_items)
+		self.recompute_material_transferred_for_manufacturing()
 
 	def refresh_material_transferred_for_manufacturing(self):
 		"""Recompute material_transferred_for_manufacturing only, without touching per-row
@@ -157,26 +159,29 @@ class RequiredItemsService:
 		"""
 		if self.doc.skip_transfer:
 			return
-		transferred_items = self._material_transfer_qty_by_item(is_return=0)
-		self.recompute_material_transferred_for_manufacturing(transferred_items)
+		self.recompute_material_transferred_for_manufacturing()
 
-	def recompute_material_transferred_for_manufacturing(self, transferred_items):
-		"""Set material_transferred_for_manufacturing based on actual item-level transfers, not fg_completed_qty."""
+	def recompute_material_transferred_for_manufacturing(self):
+		"""Set material_transferred_for_manufacturing to the finished-good qty covered by net
+		item-level transfers (transfers minus returns), capped at the transfer allowance.
+		Falls back to the claimed SUM(fg_completed_qty) when coverage is unmeasurable.
+		"""
 		# Job Card transfers use the minimum completed quantity across operations.
 		if self.doc.operations and self.doc.transfer_material_against == "Job Card":
 			return
 
-		# When fg_completed_qty > 0 (direct stock entries, excess transfer), preserve the
-		# SUM(fg_completed_qty) approach so excess-transfer tracking works correctly.
-		sum_fg_completed_qty = StatusService(self.doc).get_transferred_or_manufactured_qty(
-			"Material Transfer for Manufacture", "material_transferred_for_manufacturing"
-		)
-		if sum_fg_completed_qty:
-			self.doc.db_set("material_transferred_for_manufacturing", sum_fg_completed_qty)
-			return
+		covered_qty = self._transfer_covered_qty()
+		if covered_qty is None:
+			covered_qty = StatusService(self.doc).get_transferred_or_manufactured_qty(
+				"Material Transfer for Manufacture", "material_transferred_for_manufacturing"
+			)
+		self.doc.db_set("material_transferred_for_manufacturing", covered_qty)
 
-		# Pick list flow sets fg_completed_qty=0; use min-fraction of actual item transfers
-		# so partial availability does not prematurely mark the work order as fully transferred.
+	def _transfer_covered_qty(self):
+		"""Finished-good qty covered by net transferred raw materials, None when unmeasurable.
+		Fractions marginally below a landmark (full transfer, transfer allowance) snap to it
+		so UOM-conversion rounding losses do not leave a full transfer marginally short.
+		"""
 		required_by_item = {}
 		for row in self.doc.required_items:
 			if not row.include_item_in_manufacturing or flt(row.required_qty) <= 0:
@@ -184,15 +189,32 @@ class RequiredItemsService:
 			required_by_item[row.item_code] = required_by_item.get(row.item_code, 0.0) + flt(row.required_qty)
 
 		if not required_by_item:
-			return
+			return None
 
+		net_transferred = self._net_transferred_qty_by_item()
 		min_fraction = min(
-			flt(transferred_items.get(item_code) or 0) / required_qty
+			flt(net_transferred.get(item_code) or 0) / required_qty
 			for item_code, required_qty in required_by_item.items()
 		)
-		min_fraction = min(min_fraction, 1.0)
-		material_transferred = min_fraction * flt(self.doc.qty)
-		self.doc.db_set("material_transferred_for_manufacturing", material_transferred)
+		allowance = StatusService(self.doc).get_qty_allowance("Material Transfer for Manufacture")
+		allowance_fraction = 1.0 + allowance / 100.0
+		min_fraction = self._snap_fraction(min_fraction, (1.0, allowance_fraction))
+		covered_qty = min(min_fraction, allowance_fraction) * flt(self.doc.qty)
+		return flt(covered_qty, self.doc.precision("material_transferred_for_manufacturing"))
+
+	def _snap_fraction(self, fraction, landmarks):
+		for landmark in landmarks:
+			if fraction < landmark and landmark - fraction <= _FULL_TRANSFER_TOLERANCE:
+				return landmark
+		return fraction
+
+	def _net_transferred_qty_by_item(self):
+		transferred = self._material_transfer_qty_by_item(is_return=0, exclude_additional=True)
+		returned = self._material_transfer_qty_by_item(is_return=1, exclude_additional=True)
+		net = frappe._dict()
+		for item_code, qty in transferred.items():
+			net[item_code] = max(0.0, flt(qty) - flt(returned.get(item_code) or 0.0))
+		return net
 
 	def update_returned_qty(self):
 		returned_dict = self._material_transfer_qty_by_item(is_return=1)
@@ -290,7 +312,7 @@ class RequiredItemsService:
 		)
 		return frappe._dict({d.item_code: flt(d.qty) for d in query.run(as_dict=1)})
 
-	def _material_transfer_qty_by_item(self, is_return):
+	def _material_transfer_qty_by_item(self, is_return, exclude_additional=False):
 		ste = frappe.qb.DocType("Stock Entry")
 		ste_child = frappe.qb.DocType("Stock Entry Detail")
 		job_card = frappe.qb.DocType("Job Card")
@@ -311,7 +333,7 @@ class RequiredItemsService:
 				ste_child.original_item,
 				fn.Sum(ste_child.transfer_qty).as_("qty"),
 			)
-			.where(self._material_transfer_filter(ste, is_return))
+			.where(self._material_transfer_filter(ste, is_return, exclude_additional))
 			.where(fn.Coalesce(job_card.is_corrective_job_card, 0) == 0)
 			.groupby(ste_child.item_code, ste_child.original_item)
 		)
@@ -321,14 +343,16 @@ class RequiredItemsService:
 			qty_by_item[key] = (qty_by_item.get(key) or 0.0) + flt(d.qty)
 
 		if is_return:
-			return self._cap_returned_qty_to_transferred(qty_by_item)
+			return self._cap_returned_qty_to_transferred(qty_by_item, exclude_additional)
 
 		return qty_by_item
 
-	def _cap_returned_qty_to_transferred(self, returned_qty_by_item):
+	def _cap_returned_qty_to_transferred(self, returned_qty_by_item, exclude_additional=False):
 		# Work Order returns combine regular and corrective stock without a Job Card link.
 		# Cap each return at the regular transfer total so corrective quantities stay neutral.
-		transferred_qty_by_item = self._material_transfer_qty_by_item(is_return=0)
+		transferred_qty_by_item = self._material_transfer_qty_by_item(
+			is_return=0, exclude_additional=exclude_additional
+		)
 		return frappe._dict(
 			{
 				item_code: min(flt(returned_qty), flt(transferred_qty_by_item.get(item_code)))
@@ -336,13 +360,16 @@ class RequiredItemsService:
 			}
 		)
 
-	def _material_transfer_filter(self, ste, is_return):
-		return (
+	def _material_transfer_filter(self, ste, is_return, exclude_additional=False):
+		condition = (
 			(ste.docstatus == 1)
 			& (ste.work_order == self.doc.name)
 			& (ste.purpose == "Material Transfer for Manufacture")
 			& (ste.is_return == is_return)
 		)
+		if exclude_additional:
+			condition &= ste.is_additional_transfer_entry == 0
+		return condition
 
 	def update_consumed_qty_for_required_items(self):
 		"""

erpnext/manufacturing/doctype/work_order/services/status.py

@@ -10,6 +10,7 @@ callers (job cards, sales orders, production plans, patches) keep working.
 
 import frappe
 from frappe import _
+from frappe.query_builder import Case
 from frappe.query_builder.functions import IfNull, Sum
 from frappe.utils import cint, flt, get_link_to_form
 
@@ -88,9 +89,9 @@ class StatusService:
 	def update_status(self, status=None):
 		"""Update status of work order if unknown"""
 		if self.doc.docstatus == 1:
-			# Refresh material_transferred_for_manufacturing before deciding status so pick-list-
-			# driven transfers (where this qty is derived from item transfers, not fg_completed_qty)
-			# are reflected immediately, instead of only after the next status update call.
+			# Refresh material_transferred_for_manufacturing before deciding status so the
+			# item-level transfer coverage is reflected immediately, instead of only after
+			# the next status update call.
 			self.doc.refresh_material_transferred_for_manufacturing()
 
 		if self.doc.status != "Closed":
@@ -144,30 +145,20 @@ class StatusService:
 		return status
 
 	def _has_transferred_material(self):
-		"""True if any raw material was transferred against this work order via a pick list
-		or a material request (these leave material_transferred_for_manufacturing at 0 via
-		the min-fraction rule)."""
+		"""True if raw material net of returns remains transferred against this work order,
+		even when the covered qty leaves material_transferred_for_manufacturing at 0."""
 		ste = frappe.qb.DocType("Stock Entry")
 		ste_child = frappe.qb.DocType("Stock Entry Detail")
-		mr_child = frappe.qb.DocType("Stock Entry Detail")
-		# Stock Entry only carries `material_request` at the child-row level, so a Stock
-		# Entry is "MR-sourced" if *any* of its rows link back to a Material Request; once
-		# that's established, sum every row's transfer_qty, not just the linked ones (a
-		# manually appended extra row on the same entry has no material_request of its own).
-		mr_sourced_stock_entries = (
-			frappe.qb.from_(mr_child).select(mr_child.parent).where(mr_child.material_request.isnotnull())
-		)
+		signed_qty = Case().when(ste.is_return == 1, -ste_child.transfer_qty).else_(ste_child.transfer_qty)
 		qty = (
 			frappe.qb.from_(ste)
 			.inner_join(ste_child)
 			.on(ste_child.parent == ste.name)
-			.select(Sum(ste_child.transfer_qty))
+			.select(Sum(signed_qty))
 			.where(
 				(ste.work_order == self.doc.name)
 				& (ste.docstatus == 1)
 				& (ste.purpose == "Material Transfer for Manufacture")
-				& (ste.is_return == 0)
-				& (ste.pick_list.isnotnull() | ste.name.isin(mr_sourced_stock_entries))
 			)
 		).run()[0][0]
 		return flt(qty) > 0
@@ -211,8 +202,16 @@ class StatusService:
 		if self._skip_transfer_purpose(purpose):
 			return
 
+		if fieldname == "material_transferred_for_manufacturing":
+			# Owned by the net-coverage recomputation; the per-entry allowance guard runs on
+			# stock entry submit, where the submitting entry's claim is known.
+			self.doc.refresh_material_transferred_for_manufacturing()
+			self.set_process_loss_qty()
+			self._update_produced_qty_in_so()
+			return
+
 		qty = self.get_transferred_or_manufactured_qty(purpose, fieldname)
-		completed_qty = self.doc.qty + (self._qty_allowance(purpose) / 100 * self.doc.qty)
+		completed_qty = self.doc.qty + (self.get_qty_allowance(purpose) / 100 * self.doc.qty)
 		if qty > completed_qty:
 			frappe.throw(
 				_("{0} ({1}) cannot be greater than planned quantity ({2}) in Work Order {3}").format(
@@ -232,7 +231,7 @@ class StatusService:
 			and self.doc.transfer_material_against == "Job Card"
 		)
 
-	def _qty_allowance(self, purpose):
+	def get_qty_allowance(self, purpose):
 		allowance = flt(
 			frappe.db.get_single_value("Manufacturing Settings", "overproduction_percentage_for_work_order")
 		)

erpnext/manufacturing/doctype/work_order/test_work_order.py

@@ -1480,9 +1480,10 @@ class TestWorkOrder(ERPNextTestSuite):
 		del transfer_entry.get("items")[0]  # transfer only one RM
 		transfer_entry.submit()
 
-		# WO's "Material Transferred for Mfg" shows all is transferred, one RM is pending
+		# For Quantity claimed 1, but the untouched RM caps the covered qty at 0
 		work_order.reload()
-		self.assertEqual(work_order.material_transferred_for_manufacturing, 1)
+		self.assertEqual(work_order.material_transferred_for_manufacturing, 0)
+		self.assertEqual(work_order.status, "In Process")
 		self.assertEqual(work_order.required_items[0].transferred_qty, 0)
 		self.assertEqual(work_order.required_items[1].transferred_qty, 2)
 
@@ -1564,6 +1565,233 @@ class TestWorkOrder(ERPNextTestSuite):
 		work_order.reload()
 		self.assertEqual(work_order.material_transferred_for_manufacturing, 2.0)
 
+	def test_material_transferred_capped_by_actual_item_transfers(self):
+		"""A transfer entry claiming For Quantity for the whole work order while its rows
+		carry less must only count the covered qty; the remainder stays transferable."""
+		work_order = make_wo_order_test_record(planned_start_date=now(), qty=4)
+		test_stock_entry.make_stock_entry(
+			item_code="_Test Item", target="_Test Warehouse - _TC", qty=10, basic_rate=5000.0
+		)
+		test_stock_entry.make_stock_entry(
+			item_code="_Test Item Home Desktop 100", target="_Test Warehouse - _TC", qty=20, basic_rate=1000.0
+		)
+
+		transfer_entry = frappe.get_doc(
+			make_stock_entry(work_order.name, "Material Transfer for Manufacture", 4)
+		)
+		for item in transfer_entry.items:
+			if item.item_code == "_Test Item":
+				item.qty = 1
+				item.transfer_qty = 1
+		transfer_entry.submit()
+
+		work_order.reload()
+		self.assertEqual(transfer_entry.fg_completed_qty, 4.0)
+		self.assertEqual(work_order.material_transferred_for_manufacturing, 1.0)
+		self.assertEqual(work_order.status, "In Process")
+
+		remainder_entry = frappe.get_doc(
+			make_stock_entry(work_order.name, "Material Transfer for Manufacture", 3)
+		)
+		remainder_entry.submit()
+
+		work_order.reload()
+		self.assertEqual(work_order.material_transferred_for_manufacturing, 4.0)
+		self.assertEqual(work_order.required_items[0].transferred_qty, 4.0)
+
+	def test_material_transferred_counts_mixed_direct_and_pick_list_transfers(self):
+		"""Coverage from For Quantity = 0 entries (pick list / material request flow) must add
+		to coverage from claimed entries instead of being capped away by the claim."""
+		work_order = make_wo_order_test_record(planned_start_date=now(), qty=4)
+		test_stock_entry.make_stock_entry(
+			item_code="_Test Item", target="_Test Warehouse - _TC", qty=10, basic_rate=5000.0
+		)
+		test_stock_entry.make_stock_entry(
+			item_code="_Test Item Home Desktop 100", target="_Test Warehouse - _TC", qty=10, basic_rate=1000.0
+		)
+
+		direct_entry = frappe.get_doc(
+			make_stock_entry(work_order.name, "Material Transfer for Manufacture", 1)
+		)
+		direct_entry.submit()
+
+		work_order.reload()
+		self.assertEqual(work_order.material_transferred_for_manufacturing, 1.0)
+
+		pick_list_style_entry = frappe.get_doc(
+			make_stock_entry(work_order.name, "Material Transfer for Manufacture", 0)
+		)
+		pick_list_style_entry.submit()
+
+		work_order.reload()
+		self.assertEqual(work_order.material_transferred_for_manufacturing, 4.0)
+		self.assertEqual(work_order.status, "In Process")
+
+	def test_material_transferred_reduced_by_returns(self):
+		"""Returning raw material from WIP must reduce material_transferred_for_manufacturing."""
+		work_order = make_wo_order_test_record(planned_start_date=now(), qty=2)
+		test_stock_entry.make_stock_entry(
+			item_code="_Test Item", target="_Test Warehouse - _TC", qty=10, basic_rate=5000.0
+		)
+		test_stock_entry.make_stock_entry(
+			item_code="_Test Item Home Desktop 100", target="_Test Warehouse - _TC", qty=10, basic_rate=1000.0
+		)
+
+		transfer_entry = frappe.get_doc(
+			make_stock_entry(work_order.name, "Material Transfer for Manufacture", 2)
+		)
+		transfer_entry.submit()
+
+		work_order.reload()
+		self.assertEqual(work_order.material_transferred_for_manufacturing, 2.0)
+
+		return_entry = make_stock_return_entry(work_order.name)
+		return_entry.company = work_order.company
+		for row in list(return_entry.items):
+			if row.item_code != "_Test Item":
+				return_entry.remove(row)
+		return_entry.items[0].qty = 1
+		return_entry.save()
+		return_entry.submit()
+
+		work_order.reload()
+		self.assertEqual(work_order.material_transferred_for_manufacturing, 1.0)
+		self.assertEqual(work_order.status, "In Process")
+
+	def test_material_transferred_reduced_by_alternative_item_returns(self):
+		"""Returned alternative items must reduce coverage of the required item they substituted."""
+		work_order = make_wo_order_test_record(planned_start_date=now(), qty=2)
+		alternative_item = make_item(
+			"Alternative RM For WO Coverage", {"is_stock_item": 1, "stock_uom": "_Test UOM"}
+		)
+		test_stock_entry.make_stock_entry(
+			item_code=alternative_item.name, target="_Test Warehouse - _TC", qty=10, basic_rate=100.0
+		)
+		test_stock_entry.make_stock_entry(
+			item_code="_Test Item Home Desktop 100", target="_Test Warehouse - _TC", qty=10, basic_rate=1000.0
+		)
+
+		transfer_entry = frappe.get_doc(
+			make_stock_entry(work_order.name, "Material Transfer for Manufacture", 2)
+		)
+		for item in transfer_entry.items:
+			if item.item_code == "_Test Item":
+				item.item_code = alternative_item.name
+				item.original_item = "_Test Item"
+		transfer_entry.submit()
+
+		work_order.reload()
+		self.assertEqual(work_order.material_transferred_for_manufacturing, 2.0)
+
+		return_entry = make_stock_return_entry(work_order.name)
+		return_entry.company = work_order.company
+		for row in list(return_entry.items):
+			if row.item_code != alternative_item.name:
+				return_entry.remove(row)
+		self.assertEqual(return_entry.items[0].original_item, "_Test Item")
+		return_entry.items[0].qty = 1
+		return_entry.save()
+		return_entry.submit()
+
+		work_order.reload()
+		self.assertEqual(work_order.material_transferred_for_manufacturing, 1.0)
+
+	def test_material_transferred_snaps_conversion_rounding_losses(self):
+		"""A full transfer whose rows come out marginally short from UOM-conversion rounding
+		must count as fully transferred instead of storing values like 172.4789 for 172.5."""
+		work_order = make_wo_order_test_record(planned_start_date=now(), qty=2000)
+		test_stock_entry.make_stock_entry(
+			item_code="_Test Item", target="_Test Warehouse - _TC", qty=5000, basic_rate=5000.0
+		)
+		test_stock_entry.make_stock_entry(
+			item_code="_Test Item Home Desktop 100",
+			target="_Test Warehouse - _TC",
+			qty=5000,
+			basic_rate=1000.0,
+		)
+
+		required_qty = {row.item_code: flt(row.required_qty) for row in work_order.required_items}
+		transfer_entry = frappe.get_doc(
+			make_stock_entry(work_order.name, "Material Transfer for Manufacture", 0)
+		)
+		for item in transfer_entry.items:
+			item.qty = required_qty[item.item_code] * 0.9995
+			item.transfer_qty = item.qty
+		transfer_entry.submit()
+
+		work_order.reload()
+		self.assertEqual(work_order.material_transferred_for_manufacturing, 2000.0)
+
+	def test_return_attribution_when_item_doubles_as_alternative(self):
+		"""An item transferred both for its own requirement and as an alternative for
+		another requirement must return per attribution, not under one original_item."""
+		work_order = make_wo_order_test_record(planned_start_date=now(), qty=2)
+		test_stock_entry.make_stock_entry(
+			item_code="_Test Item", target="_Test Warehouse - _TC", qty=10, basic_rate=5000.0
+		)
+
+		transfer_entry = frappe.get_doc(
+			make_stock_entry(work_order.name, "Material Transfer for Manufacture", 2)
+		)
+		for item in transfer_entry.items:
+			if item.item_code == "_Test Item Home Desktop 100":
+				item.item_code = "_Test Item"
+				item.original_item = "_Test Item Home Desktop 100"
+		transfer_entry.submit()
+
+		work_order.reload()
+		self.assertEqual(work_order.material_transferred_for_manufacturing, 2.0)
+
+		return_entry = make_stock_return_entry(work_order.name)
+		return_entry.company = work_order.company
+		rows_by_attribution = {row.original_item: row for row in return_entry.items}
+		self.assertEqual(set(rows_by_attribution), {None, "_Test Item Home Desktop 100"})
+		self.assertEqual(rows_by_attribution[None].qty, 2)
+		self.assertEqual(rows_by_attribution["_Test Item Home Desktop 100"].qty, 4)
+
+		return_entry.remove(rows_by_attribution[None])
+		return_entry.items[0].qty = 2
+		return_entry.save()
+		return_entry.submit()
+
+		work_order.reload()
+		returned_by_item = {row.item_code: row.returned_qty for row in work_order.required_items}
+		self.assertEqual(returned_by_item["_Test Item"], 0)
+		self.assertEqual(returned_by_item["_Test Item Home Desktop 100"], 2)
+		self.assertEqual(work_order.material_transferred_for_manufacturing, 1.0)
+
+	def test_return_after_consumption_distributes_across_attributions(self):
+		"""Manufacture consumption carries no original_item; it must drain attribution
+		buckets in transfer order so the return entry reflects what remains."""
+		frappe.db.set_single_value(
+			"Manufacturing Settings",
+			"backflush_raw_materials_based_on",
+			"Material Transferred for Manufacture",
+		)
+		work_order = make_wo_order_test_record(planned_start_date=now(), qty=2)
+		test_stock_entry.make_stock_entry(
+			item_code="_Test Item", target="_Test Warehouse - _TC", qty=10, basic_rate=5000.0
+		)
+
+		transfer_entry = frappe.get_doc(
+			make_stock_entry(work_order.name, "Material Transfer for Manufacture", 2)
+		)
+		for item in transfer_entry.items:
+			if item.item_code == "_Test Item Home Desktop 100":
+				item.item_code = "_Test Item"
+				item.original_item = "_Test Item Home Desktop 100"
+		transfer_entry.submit()
+
+		manufacture_entry = frappe.get_doc(make_stock_entry(work_order.name, "Manufacture", 1))
+		raw_material_rows = [row for row in manufacture_entry.items if row.s_warehouse]
+		self.assertEqual(sorted(row.qty for row in raw_material_rows), [1.0, 2.0])
+		manufacture_entry.submit()
+
+		return_entry = make_stock_return_entry(work_order.name)
+		self.assertEqual(len(return_entry.items), 1)
+		self.assertEqual(return_entry.items[0].original_item, "_Test Item Home Desktop 100")
+		self.assertEqual(return_entry.items[0].qty, 3)
+
 	def test_status_in_process_when_only_one_required_item_transferred(self):
 		"""Stock Entry created from a Pick List that picked only one of the required items:
 		min-fraction keeps material_transferred_for_manufacturing at 0, but the work order must

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

@@ -668,7 +668,11 @@ class ManufactureStockEntry(BaseManufactureStockEntry):
 		qty = row.qty if is_return else (flt(row.qty) * flt(self.doc.fg_completed_qty)) / pending_qty_to_mfg
 		item_args["qty"] = ceil_qty_if_uom_has_whole_number(qty, row.uom)
 		item_args["transfer_qty"] = item_args["qty"]
+		if not flt(item_args["qty"], frappe.get_precision("Stock Entry Detail", "qty")):
+			return
 		if is_return:
+			if row.get("original_item"):
+				item_args["original_item"] = row.original_item
 			item_args["s_warehouse"], item_args["t_warehouse"] = row.s_warehouse, row.t_warehouse
 		else:
 			item_args["t_warehouse"], item_args["s_warehouse"] = None, row.warehouse
@@ -758,7 +762,7 @@ class ManufactureStockEntry(BaseManufactureStockEntry):
 	def add_materials_from_transfer(self):
 		for row in self._transfer_entries:
 			row.warehouse = row.t_warehouse
-			key = (row.item_code, row.warehouse)
+			key = (row.item_code, row.warehouse, row.original_item or None)
 			if key not in self.available_materials:
 				self.available_materials[key] = frappe._dict(row)
 			else:
@@ -788,20 +792,49 @@ class ManufactureStockEntry(BaseManufactureStockEntry):
 	def remove_consumed_materials_from_available(self):
 		for row in self._consumption_entries:
 			row.warehouse = row.s_warehouse
-			key = (row.item_code, row.warehouse)
-			self.available_materials[key].qty -= row.qty
+			buckets = self._get_available_buckets(row.item_code, row.warehouse)
+			self._deduct_consumed_qty(buckets, flt(row.qty))
 			if row.serial_and_batch_bundle:
-				self._deduct_consumed_serial_batch(key, row.serial_and_batch_bundle)
+				self._deduct_consumed_serial_batch(buckets, row.serial_and_batch_bundle)
 
-	def _deduct_consumed_serial_batch(self, key, sabb_name):
+	def _get_available_buckets(self, item_code, warehouse):
+		return [
+			bucket
+			for key, bucket in self.available_materials.items()
+			if key[0] == item_code and key[1] == warehouse
+		]
+
+	def _deduct_consumed_qty(self, buckets, consumed_qty):
+		for bucket in buckets[:-1]:
+			deducted = min(max(flt(bucket.qty), 0.0), consumed_qty)
+			bucket.qty -= deducted
+			consumed_qty -= deducted
+		buckets[-1].qty -= consumed_qty
+
+	def _deduct_consumed_serial_batch(self, buckets, sabb_name):
 		_details = self.get_sabb_details(sabb_name)
 		if _details.serial_nos:
-			for sn in _details.serial_nos:
-				self.available_materials[key].serial_nos.remove(sn)
+			for serial_no in _details.serial_nos:
+				self._get_serial_no_bucket(buckets, serial_no).serial_nos.remove(serial_no)
 		elif _details.batches:
 			for batch_no, qty in _details.batches.items():
-				# qty is negative, so add instead of subtract
-				self.available_materials[key].batches[batch_no] += qty
+				self._deduct_consumed_batch_qty(buckets, batch_no, -qty)
+
+	def _get_serial_no_bucket(self, buckets, serial_no):
+		for bucket in buckets:
+			if bucket.serial_nos and serial_no in bucket.serial_nos:
+				return bucket
+		return buckets[-1]
+
+	def _deduct_consumed_batch_qty(self, buckets, batch_no, consumed_qty):
+		holders = [bucket for bucket in buckets if bucket.batches and batch_no in bucket.batches]
+		if not holders:
+			holders = buckets[-1:]
+		for bucket in holders[:-1]:
+			deducted = min(max(flt(bucket.batches[batch_no]), 0.0), consumed_qty)
+			bucket.batches[batch_no] -= deducted
+			consumed_qty -= deducted
+		holders[-1].batches[batch_no] -= consumed_qty
 
 	def add_additional_cost(self):
 		if not self.wo_doc:

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

@@ -382,6 +382,7 @@ class MaterialTransferForManufactureStockEntry(BaseMaterialTransferStockEntry):
 			if self.doc.fg_completed_qty:
 				if self.doc.docstatus == 1:
 					self.wo_doc.add_additional_items(self.doc)
+					self._validate_transfer_within_allowance()
 				else:
 					self.wo_doc.remove_additional_items(self.doc)
 
@@ -391,6 +392,40 @@ class MaterialTransferForManufactureStockEntry(BaseMaterialTransferStockEntry):
 			if not self.wo_doc.operations:
 				self.wo_doc.set_actual_dates()
 
+	def _validate_transfer_within_allowance(self):
+		"""Reject a transfer whose For Quantity, on top of the effective qty already
+		transferred, exceeds the planned qty plus the transfer allowance. The projection
+		is bounded by the claim sum, which already excludes corrective job card and
+		additional transfer entries, so claim-less coverage never consumes the budget."""
+		from erpnext.manufacturing.doctype.work_order.services.status import StatusService
+		from erpnext.manufacturing.doctype.work_order.work_order import StockOverProductionError
+
+		if self.doc.is_return or self.doc.is_additional_transfer_entry:
+			return
+		if self.wo_doc.track_semi_finished_goods:
+			return
+		if self.wo_doc.operations and self.wo_doc.transfer_material_against == "Job Card":
+			return
+
+		status_service = StatusService(self.wo_doc)
+		allowance = status_service.get_qty_allowance("Material Transfer for Manufacture")
+		allowed_qty = flt(self.wo_doc.qty) * (1.0 + allowance / 100.0)
+		transferred_qty = flt(self.wo_doc.material_transferred_for_manufacturing)
+		claimed_qty = status_service.get_transferred_or_manufactured_qty(
+			"Material Transfer for Manufacture", "material_transferred_for_manufacturing"
+		)
+		projected_qty = min(transferred_qty + flt(self.doc.fg_completed_qty), claimed_qty)
+		precision = self.wo_doc.precision("material_transferred_for_manufacturing")
+		if flt(projected_qty, precision) <= flt(allowed_qty, precision):
+			return
+
+		frappe.throw(
+			_(
+				"For Quantity ({0}) with the already transferred quantity ({1}) cannot be greater than allowed quantity ({2}) in Work Order {3}"
+			).format(flt(self.doc.fg_completed_qty), transferred_qty, allowed_qty, self.wo_doc.name),
+			StockOverProductionError,
+		)
+
 
 class MaterialRequestStockEntry(BaseMaterialTransferStockEntry):
 	def before_validate(self):