fix(manufacturing): preserve attribution through consumption (#58146)

* fix(manufacturing): preserve attribution through consumption

* fix(manufacturing): preserve consumed quantity matching

* fix(manufacturing): assign consumption once

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

@@ -666,7 +666,8 @@ def _consumed_qty_filter(stock_entry, stock_entry_detail, work_order, item_code)
 		& (stock_entry.purpose.isin(["Manufacture", "Material Consumption for Manufacture"]))
 		& (stock_entry.docstatus == 1)
 		& (stock_entry_detail.s_warehouse.isnotnull())
-		& ((stock_entry_detail.item_code == item_code) | (stock_entry_detail.original_item == item_code))
+		# An attributed row belongs to its original requirement, not both item codes.
+		& (fn.Coalesce(stock_entry_detail.original_item, stock_entry_detail.item_code) == item_code)
 	)
 
 

erpnext/manufacturing/doctype/work_order/test_work_order.py

@@ -1690,9 +1690,7 @@ class TestWorkOrder(ERPNextTestSuite):
 			terminal_work_order.reload()
 			self.assertEqual(terminal_work_order.material_transferred_for_manufacturing, 1.99)
 
-	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."""
+	def _make_shared_alternative_transfer(self):
 		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
@@ -1706,6 +1704,11 @@ class TestWorkOrder(ERPNextTestSuite):
 				item.item_code = "_Test Item"
 				item.original_item = "_Test Item Home Desktop 100"
 		transfer_entry.submit()
+		return work_order
+
+	def test_return_attribution_when_item_doubles_as_alternative(self):
+		"""An item transferred for itself and as an alternative must return per requirement."""
+		work_order = self._make_shared_alternative_transfer()
 
 		work_order.reload()
 		self.assertEqual(work_order.material_transferred_for_manufacturing, 2.0)
@@ -1730,32 +1733,22 @@ class TestWorkOrder(ERPNextTestSuite):
 	@ERPNextTestSuite.change_settings(
 		"Manufacturing Settings", {"backflush_raw_materials_based_on": "Material Transferred for Manufacture"}
 	)
-	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."""
-		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()
+	def test_return_after_consumption_preserves_attribution(self):
+		work_order = self._make_shared_alternative_transfer()
 
 		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])
+		consumed_by_attribution = {
+			row.original_item: row.qty for row in manufacture_entry.items if row.s_warehouse
+		}
+		self.assertEqual(consumed_by_attribution, {None: 1.0, "_Test Item Home Desktop 100": 2.0})
 		manufacture_entry.submit()
+		work_order.reload()
+		consumed_by_item = {row.item_code: row.consumed_qty for row in work_order.required_items}
+		self.assertEqual(consumed_by_item, {"_Test Item": 1.0, "_Test Item Home Desktop 100": 2.0})
 
 		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)
+		available_by_attribution = {row.original_item: row.qty for row in return_entry.items}
+		self.assertEqual(available_by_attribution, {None: 1.0, "_Test Item Home Desktop 100": 2.0})
 
 	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:

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

@@ -664,6 +664,8 @@ class ManufactureStockEntry(BaseManufactureStockEntry):
 
 	def _append_transfer_based_rm(self, row, pending_qty_to_mfg):
 		item_args = self.get_item_dict(row)
+		if row.original_item:
+			item_args["original_item"] = row.original_item
 		is_return = self.doc.get("is_return")
 		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)
@@ -671,7 +673,6 @@ class ManufactureStockEntry(BaseManufactureStockEntry):
 		if not flt(item_args["qty"], frappe.get_precision("Stock Entry Detail", "qty")):
 			return
 		if is_return:
-			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
@@ -755,6 +756,8 @@ class ManufactureStockEntry(BaseManufactureStockEntry):
 				& (stock_entry.purpose == "Material Transfer for Manufacture")
 				& (stock_entry.docstatus == 1)
 			)
+			.orderby(stock_entry.creation)
+			.orderby(stock_entry.name)
 			.orderby(stock_entry_detail.idx)
 		).run(as_dict=1)
 
@@ -791,18 +794,25 @@ class ManufactureStockEntry(BaseManufactureStockEntry):
 	def remove_consumed_materials_from_available(self):
 		for row in self._consumption_entries:
 			row.warehouse = row.s_warehouse
-			buckets = self._get_available_buckets(row.item_code, row.warehouse)
+			buckets = self._get_available_buckets(row)
 			if row.serial_and_batch_bundle:
 				self._deduct_consumed_serial_batch(buckets, row.serial_and_batch_bundle)
 			else:
 				self._deduct_consumed_qty(buckets, flt(row.qty))
 
-	def _get_available_buckets(self, item_code, warehouse):
-		return [
+	def _get_available_buckets(self, row):
+		"""Use exact attribution when present; otherwise drain the direct item first."""
+		key = (row.item_code, row.warehouse, row.original_item or None)
+		if row.original_item and key in self.available_materials:
+			return [self.available_materials[key]]
+
+		buckets = [self.available_materials[key]] if key in self.available_materials else []
+		buckets.extend(
 			bucket
-			for key, bucket in self.available_materials.items()
-			if key[0] == item_code and key[1] == warehouse
-		]
+			for material_key, bucket in self.available_materials.items()
+			if material_key[:2] == key[:2] and material_key != key
+		)
+		return buckets
 
 	def _deduct_consumed_qty(self, buckets, consumed_qty):
 		for bucket in buckets[:-1]:
@@ -821,16 +831,13 @@ class ManufactureStockEntry(BaseManufactureStockEntry):
 
 	def _deduct_consumed_serial_nos(self, buckets, serial_nos):
 		for serial_no in serial_nos:
-			bucket = self._get_serial_no_bucket(buckets, serial_no)
+			bucket = next(
+				(bucket for bucket in buckets if bucket.serial_nos and serial_no in bucket.serial_nos),
+				buckets[-1],
+			)
 			bucket.serial_nos.remove(serial_no)
 			bucket.qty -= 1
 
-	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:

erpnext/stock/doctype/stock_entry/test_stock_entry.py

@@ -1031,6 +1031,22 @@ class TestStockEntry(ERPNextTestSuite):
 		self.assertEqual([bucket.qty for bucket in batch_buckets], [2, 2])
 		self.assertEqual(batch_buckets[1].batches["BATCH-2"], 2)
 
+	def test_consumption_prefers_exact_attribution_bucket(self):
+		from erpnext.stock.doctype.stock_entry.services.manufacturing import ManufactureStockEntry
+
+		service = ManufactureStockEntry(frappe._dict())
+		alternative = frappe._dict(original_item="REQUIRED-ITEM")
+		direct = frappe._dict(original_item=None)
+		service.available_materials = frappe._dict(
+			{("ITEM", "WIP", "REQUIRED-ITEM"): alternative, ("ITEM", "WIP", None): direct}
+		)
+
+		legacy_row = frappe._dict(item_code="ITEM", warehouse="WIP", original_item=None)
+		self.assertEqual(service._get_available_buckets(legacy_row), [direct, alternative])
+
+		attributed_row = frappe._dict(item_code="ITEM", warehouse="WIP", original_item="REQUIRED-ITEM")
+		self.assertEqual(service._get_available_buckets(attributed_row), [alternative])
+
 	@ERPNextTestSuite.change_settings("Manufacturing Settings", {"material_consumption": 1})
 	def test_work_order_manufacture_with_material_consumption(self):
 		from erpnext.manufacturing.doctype.work_order.mapper import (