fix(manufacturing): account for process loss in Production Plan Work Orders (#58799)

erpnext/manufacturing/doctype/job_card/job_card.py

@@ -35,6 +35,9 @@ from erpnext.manufacturing.doctype.bom.bom import (
 from erpnext.manufacturing.doctype.manufacturing_settings.manufacturing_settings import (
 	get_mins_between_operations,
 )
+from erpnext.manufacturing.doctype.production_plan.services.work_order_quantities import (
+	ProductionPlanWorkOrderQuantities,
+)
 from erpnext.manufacturing.doctype.workstation_type.workstation_type import get_workstations
 from erpnext.subcontracting.doctype.subcontracting_bom.subcontracting_bom import (
 	get_subcontracting_boms_for_finished_goods,
@@ -1057,6 +1060,10 @@ class JobCard(Document):
 		if not self.operation_id:
 			return
 
+		work_order = frappe.get_doc("Work Order", self.work_order)
+		if work_order.production_plan:
+			ProductionPlanWorkOrderQuantities(work_order.production_plan).lock_plan_row(work_order)
+
 		job_cards = frappe.get_all(
 			"Job Card",
 			filters={
@@ -1071,14 +1078,9 @@ class JobCard(Document):
 		completed_qty = sum(max(flt(row.manufactured_qty), flt(row.total_completed_qty)) for row in job_cards)
 
 		frappe.db.set_value("Work Order Operation", self.operation_id, "completed_qty", completed_qty)
-		if (
-			self.finished_good
-			and frappe.get_cached_value("Work Order", self.work_order, "production_item")
-			== self.finished_good
-		):
-			_wo_doc = frappe.get_doc("Work Order", self.work_order)
-			_wo_doc.db_set("produced_qty", sum(flt(row.manufactured_qty) for row in job_cards))
-			_wo_doc.db_set("status", _wo_doc.get_status())
+		if self.finished_good and work_order.production_item == self.finished_good:
+			work_order.db_set("produced_qty", sum(flt(row.manufactured_qty) for row in job_cards))
+			work_order.db_set("status", work_order.get_status())
 
 	def update_corrective_in_work_order(self, wo):
 		wo.corrective_operation_cost = 0.0

erpnext/manufacturing/doctype/production_plan/production_plan.js

@@ -601,6 +601,14 @@ frappe.ui.form.on("Production Plan", {
 
 		let has_items =
 			items.filter((item) => {
+				const reference_field =
+					item.doctype === "Production Plan Item"
+						? "production_plan_item"
+						: "production_plan_sub_assembly_item";
+				const pending_qty = frm.doc.__onload?.pending_work_order_qty?.[reference_field]?.[item.name];
+				if (pending_qty !== undefined) {
+					return pending_qty > 0;
+				}
 				if (item.planned_qty) {
 					return item.planned_qty > item.ordered_qty;
 				} else {

erpnext/manufacturing/doctype/production_plan/production_plan.py

@@ -47,6 +47,9 @@ from erpnext.manufacturing.doctype.production_plan.services.sub_assembly import
 from erpnext.manufacturing.doctype.production_plan.services.work_order_planning import (
 	WorkOrderCreationService,
 )
+from erpnext.manufacturing.doctype.production_plan.services.work_order_quantities import (
+	ProductionPlanWorkOrderQuantities,
+)
 from erpnext.stock.utils import get_or_make_bin
 from erpnext.utilities.transaction_base import validate_uom_is_integer
 
@@ -133,6 +136,11 @@ class ProductionPlan(Document):
 			"enable_stock_reservation",
 			frappe.db.get_single_value("Stock Settings", "enable_stock_reservation"),
 		)
+		if self.docstatus == 1:
+			self.set_onload(
+				"pending_work_order_qty",
+				ProductionPlanWorkOrderQuantities(self.name).get_pending_quantities(self),
+			)
 
 	def on_discard(self):
 		self.db_set("status", "Cancelled")

erpnext/manufacturing/doctype/production_plan/services/work_order_planning.py

@@ -4,12 +4,16 @@
 """Work Order / subcontract PO creation from a Production Plan (extracted from production_plan.py)."""
 
 from collections import defaultdict
+from functools import cached_property
 
 import frappe
 from frappe import _, msgprint
 from frappe.utils import flt, get_filtered_list_link, getdate, nowdate
 
 from erpnext.manufacturing.doctype.production_plan.services.planning_queries import set_default_warehouses
+from erpnext.manufacturing.doctype.production_plan.services.work_order_quantities import (
+	ProductionPlanWorkOrderQuantities,
+)
 
 _SUB_ASSEMBLY_WO_FIELDS = [
 	"production_item",
@@ -42,17 +46,15 @@ class WorkOrderCreationService:
 		item_dict = {}
 		for d in self.doc.po_items:
 			item_details = self._production_item_details(d, bom_warehouse_map)
-			if self.doc.get_items_from == "Material Request":
-				item_details["qty"] = d.planned_qty
-				key = (d.item_code, d.material_request_item, d.warehouse, d.planned_start_date)
-				item_dict[key] = item_details
-			else:
-				key = self._production_item_key(d)
-				existing = flt(item_dict.get(key, {}).get("qty"))
-				item_details["qty"] = existing + (flt(d.planned_qty) - flt(d.ordered_qty))
-				item_dict[key] = item_details
+			item_details["qty"] = self.pending_quantities["production_plan_item"][d.name]
+			# A Work Order can reference only one Production Plan row.
+			item_dict[d.name] = item_details
 		return item_dict
 
+	@cached_property
+	def pending_quantities(self):
+		return ProductionPlanWorkOrderQuantities(self.doc.name).get_pending_quantities(self.doc)
+
 	def get_bom_source_warehouse_map(self, rows):
 		bom_names = {row.bom_no for row in rows if row.bom_no}
 		if not bom_names:
@@ -90,16 +92,10 @@ class WorkOrderCreationService:
 			details["project"] = frappe.get_cached_value("Sales Order", d.sales_order, "project")
 		return details
 
-	def _production_item_key(self, d):
-		if not d.sales_order:
-			return (d.name, d.item_code, d.warehouse, d.planned_start_date)
-		if self.doc.combine_items:
-			return (d.item_code, d.sales_order, d.warehouse, d.planned_start_date)
-		return (d.item_code, d.sales_order, d.sales_order_item, d.warehouse, d.planned_start_date)
-
 	def make_work_order(self):
 		from erpnext.manufacturing.doctype.work_order.work_order import get_default_warehouse
 
+		self.doc.reload()
 		wo_list, po_list = [], []
 		subcontracted_po = {}
 		default_warehouses = get_default_warehouse(self.doc.company)
@@ -139,9 +135,6 @@ class WorkOrderCreationService:
 				wo_list.append(work_order)
 
 	def _sub_assembly_work_order(self, row, default_warehouses, bom_warehouse_map):
-		if flt(row.qty) <= flt(row.ordered_qty):
-			return None
-
 		work_order_data = {
 			"source_warehouse": bom_warehouse_map.get(row.bom_no),
 			"wip_warehouse": default_warehouses.get("wip_warehouse"),
@@ -159,7 +152,7 @@ class WorkOrderCreationService:
 			if row.get(field):
 				wo_data[field] = row.get(field)
 
-		wo_data["qty"] = flt(row.get("qty")) - flt(row.get("ordered_qty"))
+		wo_data["qty"] = self.pending_quantities["production_plan_sub_assembly_item"][row.name]
 		wo_data.update(
 			{
 				"use_multi_level_bom": 0,

erpnext/manufacturing/doctype/production_plan/services/work_order_quantities.py

@@ -0,0 +1,131 @@
+# Copyright (c) 2026, Frappe Technologies Pvt. Ltd. and Contributors
+# License: GNU General Public License v3. See license.txt
+
+from collections import defaultdict
+
+import frappe
+from frappe import _
+from frappe.utils import flt, get_link_to_form
+
+
+class ProductionPlanWorkOrderQuantities:
+	"""Count submitted Work Orders after recorded process loss, independently for each plan row."""
+
+	def __init__(self, production_plan):
+		self.production_plan = production_plan
+
+	def validate_work_order(self, work_order, *, process_loss_qty=0):
+		from erpnext.manufacturing.doctype.work_order.work_order import OverProductionError
+
+		row = self.lock_plan_row(work_order)
+
+		committed = self.get_committed_quantities(
+			exclude_work_order=work_order.name,
+			reference_field=row.reference_field,
+			reference_name=row.name,
+			for_update=True,
+		)[row.reference_field].get(row.name, 0)
+		allowance = flt(
+			frappe.db.get_single_value("Manufacturing Settings", "overproduction_percentage_for_work_order")
+		)
+		precision = work_order.precision("qty")
+		maximum_qty = flt(
+			flt(row.planned_qty) * (1 + allowance / 100) - committed + flt(process_loss_qty), precision
+		)
+		if flt(work_order.qty, precision) > maximum_qty:
+			frappe.throw(
+				_(
+					"Row {0} in {1} {2}: Work Order quantity {3} exceeds the remaining allowed quantity {4}."
+				).format(
+					row.idx,
+					_(row.doctype),
+					get_link_to_form("Production Plan", self.production_plan),
+					flt(work_order.qty, precision),
+					max(0, maximum_qty),
+				),
+				OverProductionError,
+				title=_("Production Plan Quantity Exceeded"),
+			)
+
+	def lock_plan_row(self, work_order):
+		if work_order.production_plan_item and work_order.production_plan_sub_assembly_item:
+			frappe.throw(_("Work Order must reference only one Production Plan row."))
+
+		if work_order.production_plan_sub_assembly_item:
+			reference_field = "production_plan_sub_assembly_item"
+			row_doctype, qty_field = "Production Plan Sub Assembly Item", "qty"
+		else:
+			reference_field = "production_plan_item"
+			row_doctype, qty_field = "Production Plan Item", "planned_qty"
+
+		reference_name = work_order.get(reference_field)
+		# Serialize submissions and loss reversals. The submit rollup updates this row.
+		row = (
+			frappe.db.get_value(
+				row_doctype,
+				{"name": reference_name, "parent": self.production_plan},
+				["name", "idx", f"{qty_field} as planned_qty"],
+				as_dict=True,
+				for_update=True,
+			)
+			if reference_name
+			else None
+		)
+		if not row:
+			frappe.throw(
+				_("Work Order must reference a row in Production Plan {0}.").format(
+					get_link_to_form("Production Plan", self.production_plan)
+				)
+			)
+
+		row.reference_field = reference_field
+		row.doctype = row_doctype
+		return row
+
+	def get_pending_quantities(self, plan):
+		committed = self.get_committed_quantities()
+		precision = frappe.get_precision("Work Order", "qty")
+		pending = {}
+		for table, reference_field, qty_field in (
+			("po_items", "production_plan_item", "planned_qty"),
+			("sub_assembly_items", "production_plan_sub_assembly_item", "qty"),
+		):
+			pending[reference_field] = {
+				row.name: max(
+					0, flt(flt(row.get(qty_field)) - committed[reference_field].get(row.name, 0), precision)
+				)
+				for row in plan.get(table)
+				if table == "po_items" or row.type_of_manufacturing == "In House"
+			}
+		return pending
+
+	def get_committed_quantities(
+		self, exclude_work_order=None, reference_field=None, reference_name=None, for_update=False
+	):
+		filters = {"production_plan": self.production_plan, "docstatus": 1}
+		if exclude_work_order:
+			filters["name"] = ("!=", exclude_work_order)
+		if reference_field:
+			filters[reference_field] = reference_name
+
+		# Read rows instead of an aggregate so MariaDB uses a current locking read on submit.
+		work_orders = frappe.qb.get_query(
+			"Work Order",
+			fields=["production_plan_item", "production_plan_sub_assembly_item", "qty", "process_loss_qty"],
+			filters=filters,
+			for_update=for_update,
+			order_by="name",
+		).run(as_dict=True)
+		quantities = {
+			"production_plan_item": defaultdict(float),
+			"production_plan_sub_assembly_item": defaultdict(float),
+		}
+		for work_order in work_orders:
+			field = (
+				"production_plan_sub_assembly_item"
+				if work_order.production_plan_sub_assembly_item
+				else "production_plan_item"
+			)
+			if work_order.get(field):
+				quantities[field][work_order[field]] += flt(work_order.qty) - flt(work_order.process_loss_qty)
+		return quantities

erpnext/manufacturing/doctype/production_plan/test_production_plan.py

@@ -1160,12 +1160,12 @@ class TestProductionPlan(ERPNextTestSuite):
 	def test_multiple_work_order_for_production_plan_item(self):
 		"Test producing Prod Plan (making WO) in parts."
 
-		def create_work_order(item, pln, qty):
+		def create_work_order(pln, qty):
 			# Get Production Items
 			items_data = pln.get_production_items()
 
 			# Update qty
-			items_data[(pln.po_items[0].name, item, None, pln.po_items[0].planned_start_date)]["qty"] = qty
+			items_data[pln.po_items[0].name]["qty"] = qty
 
 			# Create and Submit Work Order for each item in items_data
 			for _key, item in items_data.items():
@@ -1193,17 +1193,17 @@ class TestProductionPlan(ERPNextTestSuite):
 		wo_list = []
 
 		# Create and Submit 1st Work Order for 3 qty
-		create_work_order(item, pln, 3)
+		create_work_order(pln, 3)
 		pln.reload()
 		self.assertEqual(pln.po_items[0].ordered_qty, 3)
 
 		# Create and Submit 2nd Work Order for 2 qty
-		create_work_order(item, pln, 2)
+		create_work_order(pln, 2)
 		pln.reload()
 		self.assertEqual(pln.po_items[0].ordered_qty, 5)
 
 		# Overproduction
-		self.assertRaises(OverProductionError, create_work_order, item=item, pln=pln, qty=2)
+		self.assertRaises(OverProductionError, create_work_order, pln=pln, qty=2)
 
 		# Cancel 1st Work Order
 		wo1 = frappe.get_doc("Work Order", wo_list[0])
@@ -1384,8 +1384,11 @@ class TestProductionPlan(ERPNextTestSuite):
 		make_bom(item=fg_item, raw_materials=[sub_assembly_item], rm_qty=4)
 
 		# Step - 1: Create Production Plan
-		pln = create_production_plan(item_code=fg_item, planned_qty=5, skip_getting_mr_items=1)
+		pln = create_production_plan(
+			item_code=fg_item, planned_qty=5, skip_getting_mr_items=1, do_not_submit=1
+		)
 		pln.get_sub_assembly_items()
+		pln.submit()
 
 		# Step - 2: Create Work Orders
 		pln.make_work_order()

erpnext/manufacturing/doctype/production_plan/test_work_order_quantities.py

@@ -0,0 +1,413 @@
+# Copyright (c) 2026, Frappe Technologies Pvt. Ltd. and Contributors
+# License: GNU General Public License v3. See license.txt
+
+from contextlib import contextmanager
+from unittest.mock import patch
+
+import frappe
+
+from erpnext.manufacturing.doctype.operation.test_operation import make_operation
+from erpnext.manufacturing.doctype.production_plan.test_production_plan import (
+	create_production_plan,
+	make_bom,
+)
+from erpnext.manufacturing.doctype.work_order.mapper import make_stock_entry as make_se_from_wo
+from erpnext.manufacturing.doctype.work_order.work_order import (
+	OverProductionError,
+	close_work_order,
+	stop_unstop,
+)
+from erpnext.manufacturing.doctype.workstation.test_workstation import make_workstation
+from erpnext.selling.doctype.sales_order.test_sales_order import make_sales_order
+from erpnext.stock.doctype.item.test_item import make_item
+from erpnext.stock.doctype.stock_entry.test_stock_entry import make_stock_entry
+from erpnext.tests.utils import ERPNextTestSuite
+
+REFERENCE_FIELDS = ("production_plan_item", "production_plan_sub_assembly_item")
+
+
+class TestProductionPlanWorkOrderQuantities(ERPNextTestSuite):
+	def setUp(self):
+		self.warehouse = "_Test Warehouse - _TC"
+		self.raw_material, self.sub_assembly, self.finished_good = (
+			make_item(properties={"is_stock_item": 1, "stock_uom": "Kg", "valuation_rate": 10}).name
+			for _ in range(3)
+		)
+		for item, material in (
+			(self.sub_assembly, self.raw_material),
+			(self.finished_good, self.sub_assembly),
+		):
+			make_bom(item=item, raw_materials=[material], process_loss_percentage=10)
+		frappe.db.set_single_value("Manufacturing Settings", "overproduction_percentage_for_work_order", 0)
+
+	def test_quantity_limit_on_submit(self):
+		plan = self.make_plan()
+		for field in REFERENCE_FIELDS:
+			with self.subTest(field=field):
+				first = self.create_work_order(plan, field)
+				first.qty = 50
+				first.submit()
+				second = self.create_work_order(plan, field)
+				self.assertEqual(second.qty, 50)
+				self.assert_overproduction(second, 60)
+				second.qty = 50
+				second.submit()
+				self.assert_pending_qty(plan, field, 0)
+
+	def test_recorded_loss_creates_replacement(self):
+		plan = self.make_plan()
+		for field in REFERENCE_FIELDS:
+			with self.subTest(field=field):
+				first = self.create_work_order(plan, field)
+				first.submit()
+				self.assert_pending_qty(plan, field, 0)
+				manufacture = self.manufacture_with_loss(first)
+				first.reload()
+				self.assertEqual(first.process_loss_qty, 10)
+				self.assertEqual(first.produced_qty, 90)
+				self.assert_pending_qty(plan, field, 10)
+
+				replacement = self.create_work_order(plan, field)
+				self.assertEqual(replacement.qty, 10)
+				self.assert_overproduction(replacement, 11)
+				replacement.qty = 10
+				replacement.submit()
+				self.assert_pending_qty(plan, field, 0)
+				row = self.plan_row(plan, field)
+				self.assertEqual(row.ordered_qty, 110)
+
+				self.assert_loss_reversal_blocked(manufacture)
+				first.reload()
+				self.assertEqual(first.process_loss_qty, 10)
+				self.assertEqual(first.produced_qty, 90)
+				self.assert_pending_qty(plan, field, 0)
+				replacement.cancel()
+				manufacture.cancel()
+				self.assertEqual(first.reload().process_loss_qty, 0)
+				first.reload().cancel()
+				self.assert_pending_qty(plan, field, 100)
+
+	def test_loss_reversal_with_draft_replacement(self):
+		plan = self.make_plan()
+		for field in REFERENCE_FIELDS:
+			with self.subTest(field=field):
+				first = self.create_work_order(plan, field)
+				first.submit()
+				manufacture = self.manufacture_with_loss(first)
+				replacement = self.create_work_order(plan, field)
+				manufacture.cancel()
+				self.assertEqual(first.reload().process_loss_qty, 0)
+				self.assert_pending_qty(plan, field, 0)
+				self.assert_overproduction(replacement, 10)
+
+	def test_partial_loss_reversal_with_overproduction_allowance(self):
+		frappe.db.set_single_value("Manufacturing Settings", "overproduction_percentage_for_work_order", 5)
+		plan = self.make_plan()
+		for field in REFERENCE_FIELDS:
+			with self.subTest(field=field):
+				first = self.create_work_order(plan, field)
+				first.submit()
+				manufactures = [self.manufacture_with_loss(first, qty=50) for _ in range(2)]
+				self.assertEqual(first.reload().process_loss_qty, 10)
+				replacement = self.create_work_order(plan, field)
+				replacement.submit()
+
+				# Retaining five units of loss keeps the net quantity at the allowed 105.
+				manufactures[1].cancel()
+				self.assertEqual(first.reload().process_loss_qty, 5)
+				self.assert_loss_reversal_blocked(manufactures[0])
+				self.assertEqual(first.reload().process_loss_qty, 5)
+				replacement.cancel()
+				manufactures[0].cancel()
+				self.assertEqual(first.reload().process_loss_qty, 0)
+
+	def test_job_card_loss_reversal_with_replacement(self):
+		self.make_bom_with_operation(self.finished_good, self.raw_material)
+		plan = self.make_plan()
+		first = self.create_work_order(plan, "production_plan_item")
+		first.submit()
+		job_card = frappe.get_last_doc("Job Card", {"work_order": first.name})
+		job_card.append("time_logs", {"from_time": "2024-05-01 08:00:00"})
+		job_card.save()
+		job_card.complete_job_card(
+			qty=90,
+			for_quantity=100,
+			pending_qty=0,
+			process_loss_qty=10,
+			end_time="2024-05-01 09:00:00",
+		)
+		job_card.reload().submit()
+		self.assertEqual(first.reload().process_loss_qty, 10)
+		make_stock_entry(item_code=self.raw_material, target=self.warehouse, qty=100, basic_rate=10)
+		manufacture = frappe.get_doc(job_card.make_stock_entry_for_semi_fg_item()).submit()
+		replacement = self.create_work_order(plan, "production_plan_item")
+		replacement.submit()
+		with self.assert_plan_locked_before_work_order_update(first):
+			manufacture.cancel()
+		job_card.reload()
+		self.assert_loss_reversal_blocked(job_card)
+		self.assertEqual(first.reload().process_loss_qty, 10)
+		self.assertEqual(first.operations[0].process_loss_qty, 10)
+		replacement.cancel()
+		job_card.cancel()
+		self.assertEqual(first.reload().process_loss_qty, 0)
+
+	def test_expected_loss_does_not_allow_extra_quantity(self):
+		plan = self.make_plan()
+		for field in REFERENCE_FIELDS:
+			with self.subTest(field=field):
+				work_order = self.create_work_order(plan, field)
+				self.assert_overproduction(work_order, 110)
+
+	def test_overproduction_allowance(self):
+		frappe.db.set_single_value("Manufacturing Settings", "overproduction_percentage_for_work_order", 10)
+		plan = self.make_plan()
+		for field in REFERENCE_FIELDS:
+			with self.subTest(field=field):
+				first = self.create_work_order(plan, field)
+				self.assertEqual(first.qty, 100)
+				first.submit()
+				self.assert_pending_qty(plan, field, 0)
+				second = self.copy_work_order(first)
+				self.assert_overproduction(second, 11)
+				second.qty = 10
+				second.submit()
+
+	def test_drafts_and_cancelled_orders_do_not_consume_quantity(self):
+		plan = self.make_plan()
+		for field in REFERENCE_FIELDS:
+			with self.subTest(field=field):
+				first = self.create_work_order(plan, field)
+				second = self.create_work_order(plan, field)
+				self.assertEqual(first.qty, second.qty)
+				first.submit()
+				self.assert_overproduction(second, 100)
+				first.cancel()
+				second.submit()
+				self.assert_pending_qty(plan, field, 0)
+
+	def test_process_loss_and_overproduction_allowance(self):
+		frappe.db.set_single_value("Manufacturing Settings", "overproduction_percentage_for_work_order", 10)
+		plan = self.make_plan()
+		for field in REFERENCE_FIELDS:
+			with self.subTest(field=field):
+				first = self.create_work_order(plan, field)
+				first.qty = 50
+				first.submit()
+				self.manufacture_with_loss(first)
+				second = self.create_work_order(plan, field)
+				self.assertEqual(second.qty, 55)
+				self.assert_overproduction(second, 66)
+				second.qty = 65
+				second.submit()
+				self.assert_pending_qty(plan, field, 0)
+
+	def test_stopped_and_closed_orders_consume_quantity(self):
+		plan = self.make_plan()
+		for field in REFERENCE_FIELDS:
+			with self.subTest(field=field):
+				first = self.create_work_order(plan, field)
+				first.submit()
+				stop_unstop(first.name, "Stopped")
+				self.assert_pending_qty(plan, field, 0)
+				stop_unstop(first.name, "Not Started")
+				close_work_order(first.name, "Closed")
+				self.assert_pending_qty(plan, field, 0)
+
+	def test_fractional_quantities(self):
+		plan = self.make_plan(qty=0.3)
+		for field in REFERENCE_FIELDS:
+			with self.subTest(field=field):
+				first = self.create_work_order(plan, field)
+				first.qty = 0.1
+				first.submit()
+				second = self.create_work_order(plan, field)
+				self.assertEqual(second.qty, 0.2)
+				second.submit()
+				self.assert_pending_qty(plan, field, 0)
+
+	def test_rows_with_same_item_are_independent(self):
+		plan = self.make_plan(submit=False)
+		sales_order = make_sales_order(item_code=self.finished_good, qty=200, warehouse=self.warehouse)
+		plan.po_items[0].sales_order = sales_order.name
+		plan.po_items[0].sales_order_item = sales_order.items[0].name
+		row = plan.po_items[0].as_dict()
+		row.pop("name")
+		row["planned_qty"] = 50
+		plan.append("po_items", row)
+		plan.submit()
+		first = self.create_work_order(plan, "production_plan_item")
+		first.submit()
+		plan.onload()
+		pending = plan.get_onload()["pending_work_order_qty"]["production_plan_item"]
+		self.assertEqual(pending[plan.po_items[0].name], 0)
+		self.assertEqual(pending[plan.po_items[1].name], 50)
+		second_name = frappe.db.get_value(
+			"Work Order", {"production_plan_item": plan.po_items[1].name, "docstatus": 0}, "name"
+		)
+		second = frappe.get_doc("Work Order", second_name)
+		self.assertEqual(second.qty, 50)
+		self.assert_overproduction(second, 51)
+
+	def test_material_request_plan_uses_remaining_quantity(self):
+		plan = self.make_plan(submit=False)
+		plan.get_items_from = "Material Request"
+		plan.submit()
+		first = self.create_work_order(plan, "production_plan_item")
+		first.qty = 50
+		first.submit()
+		second = self.create_work_order(plan, "production_plan_item")
+		self.assertEqual(second.qty, 50)
+
+	def test_reference_must_belong_to_plan(self):
+		plan = self.make_plan()
+		other_plan = self.make_plan()
+		work_order = self.create_work_order(plan, "production_plan_item")
+		work_order.production_plan = other_plan.name
+		with self.assertRaisesRegex(frappe.ValidationError, "must reference a row"):
+			work_order.submit()
+
+	def test_missing_or_ambiguous_plan_reference(self):
+		plan = self.make_plan()
+		work_order = self.create_work_order(plan, "production_plan_item")
+		work_order.production_plan_sub_assembly_item = plan.sub_assembly_items[0].name
+		with self.assertRaisesRegex(frappe.ValidationError, "only one Production Plan row"):
+			work_order.submit()
+		work_order.reload()
+		work_order.production_plan_item = None
+		with self.assertRaisesRegex(frappe.ValidationError, "must reference a row"):
+			work_order.submit()
+
+	def make_plan(self, qty=100, submit=True):
+		plan = create_production_plan(
+			item_code=self.finished_good,
+			planned_qty=qty,
+			stock_uom="Kg",
+			warehouse=self.warehouse,
+			sub_assembly_warehouse=self.warehouse,
+			skip_getting_mr_items=True,
+			do_not_submit=True,
+		)
+		plan.get_sub_assembly_items()
+		if submit:
+			plan.submit()
+		return plan
+
+	def make_bom_with_operation(self, item, material):
+		bom = make_bom(item=item, raw_materials=[material], with_operations=1, do_not_save=True)
+		bom.track_semi_finished_goods = 1
+		bom.items[0].operation_row_id = 1
+		operation = {
+			"operation": f"_Test Loss Reversal {item}",
+			"workstation": "_Test Workstation A",
+			"finished_good": item,
+			"finished_good_qty": 1,
+			"is_final_finished_good": 1,
+			"sequence_id": 1,
+			"time_in_mins": 60,
+			"source_warehouse": self.warehouse,
+			"fg_warehouse": self.warehouse,
+			"skip_material_transfer": 1,
+		}
+		make_workstation(operation)
+		make_operation(operation)
+		bom.append("operations", operation)
+		bom.insert().submit()
+
+	def create_work_order(self, plan, field):
+		plan.make_work_order()
+		name = frappe.db.get_value(
+			"Work Order",
+			{"production_plan": plan.name, field: self.plan_row(plan, field).name, "docstatus": 0},
+			"name",
+			order_by="creation desc",
+		)
+		work_order = frappe.get_doc("Work Order", name)
+		work_order.update(
+			{"skip_transfer": 1, "source_warehouse": self.warehouse, "fg_warehouse": self.warehouse}
+		)
+		return work_order
+
+	def copy_work_order(self, work_order):
+		copy = frappe.copy_doc(work_order)
+		copy.docstatus = 0
+		copy.production_plan = work_order.production_plan
+		for field in REFERENCE_FIELDS:
+			copy.set(field, work_order.get(field))
+		copy.insert()
+		return copy
+
+	def manufacture_with_loss(self, work_order, qty=None):
+		for item in work_order.required_items:
+			make_stock_entry(
+				item_code=item.item_code, target=self.warehouse, qty=item.required_qty, basic_rate=10
+			)
+		entry = frappe.get_doc(make_se_from_wo(work_order.name, "Manufacture", qty or work_order.qty))
+		entry.submit()
+		return entry
+
+	def assert_loss_reversal_blocked(self, document):
+		work_order = frappe.get_doc("Work Order", document.work_order)
+		frappe.db.savepoint("loss_reversal")
+		try:
+			with (
+				self.assert_plan_locked_before_work_order_update(work_order),
+				self.assertRaises(OverProductionError),
+			):
+				document.cancel()
+		finally:
+			# Match the request rollback after an on_cancel validation fails.
+			frappe.db.rollback(save_point="loss_reversal")
+		self.assertEqual(document.reload().docstatus, 1)
+
+	@contextmanager
+	def assert_plan_locked_before_work_order_update(self, work_order):
+		get_value, set_value = frappe.db.get_value, frappe.db.set_value
+		plan_row_locked = False
+		row_doctype = (
+			"Production Plan Sub Assembly Item"
+			if work_order.production_plan_sub_assembly_item
+			else "Production Plan Item"
+		)
+		row_name = work_order.production_plan_sub_assembly_item or work_order.production_plan_item
+
+		def get_value_with_lock_check(doctype, filters=None, *args, **kwargs):
+			nonlocal plan_row_locked
+			if doctype == "Work Order" and filters == work_order.name and kwargs.get("for_update"):
+				self.assertTrue(plan_row_locked, "Work Order locked before its Production Plan row")
+			result = get_value(doctype, filters, *args, **kwargs)
+			if (
+				doctype == row_doctype
+				and filters == {"name": row_name, "parent": work_order.production_plan}
+				and kwargs.get("for_update")
+			):
+				plan_row_locked = True
+			return result
+
+		def set_value_with_lock_check(doctype, name, *args, **kwargs):
+			if doctype == "Work Order" and name == work_order.name:
+				self.assertTrue(plan_row_locked, "Work Order updated before locking its Production Plan row")
+			return set_value(doctype, name, *args, **kwargs)
+
+		with (
+			patch.object(frappe.db, "get_value", get_value_with_lock_check),
+			patch.object(frappe.db, "set_value", set_value_with_lock_check),
+		):
+			yield
+
+	def assert_overproduction(self, work_order, qty):
+		work_order.qty = qty
+		work_order.save()
+		with self.assertRaises(OverProductionError):
+			work_order.submit()
+		work_order.reload()
+
+	def assert_pending_qty(self, plan, field, expected):
+		plan.reload()
+		plan.onload()
+		self.assertEqual(
+			plan.get_onload()["pending_work_order_qty"][field][self.plan_row(plan, field).name], expected
+		)
+
+	def plan_row(self, plan, field):
+		return plan.po_items[0] if field == "production_plan_item" else plan.sub_assembly_items[0]

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

@@ -13,6 +13,9 @@ from frappe import _
 from frappe.query_builder.functions import IfNull, Sum
 from frappe.utils import cint, flt, get_link_to_form
 
+from erpnext.manufacturing.doctype.production_plan.services.work_order_quantities import (
+	ProductionPlanWorkOrderQuantities,
+)
 from erpnext.stock.stock_balance import get_planned_qty, update_bin_qty
 
 _QTY_PURPOSES = (
@@ -193,6 +196,8 @@ class StatusService:
 		if self.doc.track_semi_finished_goods:
 			return
 
+		# Lock the plan row before any Work Order quantity update takes a row lock.
+		self.set_process_loss_qty()
 		for purpose, fieldname in _QTY_PURPOSES:
 			self._update_qty_for_purpose(purpose, fieldname)
 
@@ -223,7 +228,6 @@ class StatusService:
 			)
 
 		self.doc.db_set(fieldname, qty)
-		self.set_process_loss_qty()
 		self._update_produced_qty_in_so()
 
 	def _skip_transfer_purpose(self, purpose):
@@ -298,7 +302,20 @@ class StatusService:
 		)
 
 	def set_process_loss_qty(self):
-		self.doc.db_set("process_loss_qty", self._process_loss_qty())
+		quantities = None
+		if self.doc.docstatus == 1 and self.doc.production_plan:
+			quantities = ProductionPlanWorkOrderQuantities(self.doc.production_plan)
+			quantities.lock_plan_row(self.doc)
+
+		process_loss_qty = self._process_loss_qty()
+		if quantities:
+			previous_loss_qty = frappe.db.get_value(
+				"Work Order", self.doc.name, "process_loss_qty", for_update=True
+			)
+			if process_loss_qty < flt(previous_loss_qty):
+				# Replacement Work Orders may have consumed the recorded loss.
+				quantities.validate_work_order(self.doc, process_loss_qty=process_loss_qty)
+		self.doc.db_set("process_loss_qty", process_loss_qty)
 
 	def _process_loss_qty(self):
 		if self.doc.track_semi_finished_goods:

erpnext/manufacturing/doctype/work_order/work_order.py

@@ -19,6 +19,9 @@ from frappe.utils import (
 
 from erpnext.buying.utils import check_on_hold_or_closed_status
 from erpnext.manufacturing.doctype.bom.bom import validate_bom_no
+from erpnext.manufacturing.doctype.production_plan.services.work_order_quantities import (
+	ProductionPlanWorkOrderQuantities,
+)
 
 # Backward-compatible re-exports: these functions were moved to mapper.py.
 # Importing them here preserves existing whitelist dotted-paths
@@ -646,6 +649,8 @@ class WorkOrder(Document):
 			frappe.throw(_("Target Warehouse is required before Submit"))
 
 	def before_submit(self):
+		if self.production_plan:
+			ProductionPlanWorkOrderQuantities(self.production_plan).validate_work_order(self)
 		self.create_serial_no_batch_no()
 
 	def on_submit(self):
@@ -921,36 +926,6 @@ class WorkOrder(Document):
 				),
 			)
 
-		if self.production_plan and self.production_plan_item and not self.production_plan_sub_assembly_item:
-			qty_dict = frappe.db.get_value(
-				"Production Plan Item", self.production_plan_item, ["planned_qty", "ordered_qty"], as_dict=1
-			)
-
-			if not qty_dict:
-				return
-
-			allowance_qty = (
-				flt(
-					frappe.db.get_single_value(
-						"Manufacturing Settings", "overproduction_percentage_for_work_order"
-					)
-				)
-				/ 100
-				* qty_dict.get("planned_qty", 0)
-			)
-
-			max_qty = qty_dict.get("planned_qty", 0) + allowance_qty - qty_dict.get("ordered_qty", 0)
-
-			if max_qty <= 0:
-				frappe.throw(
-					_("Cannot produce more item for {0}").format(self.production_item), OverProductionError
-				)
-			elif self.qty > max_qty:
-				frappe.throw(
-					_("Cannot produce more than {0} items for {1}").format(max_qty, self.production_item),
-					OverProductionError,
-				)
-
 		if self.subcontracting_inward_order and self.qty > self.max_producible_qty:
 			frappe.msgprint(
 				_(