diff --git a/erpnext/manufacturing/doctype/job_card/job_card.py b/erpnext/manufacturing/doctype/job_card/job_card.py
index 8d2839d1827..95aaf9cf825 100644
--- a/erpnext/manufacturing/doctype/job_card/job_card.py
+++ b/erpnext/manufacturing/doctype/job_card/job_card.py
@@ -31,6 +31,9 @@ from erpnext.manufacturing.doctype.bom.bom import add_additional_cost, get_bom_i
 from erpnext.manufacturing.doctype.manufacturing_settings.manufacturing_settings import (
 	get_mins_between_operations,
 )
+from erpnext.manufacturing.doctype.production_plan.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,
@@ -990,6 +993,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={
@@ -1004,14 +1011,13 @@ 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))
+			if work_order.production_plan:
+				ProductionPlanWorkOrderQuantities(work_order.production_plan).validate_work_order(
+					work_order, process_loss_qty=work_order.process_loss_qty
+				)
+			work_order.db_set("status", work_order.get_status())
 
 	def update_corrective_in_work_order(self, wo):
 		wo.corrective_operation_cost = 0.0
diff --git a/erpnext/manufacturing/doctype/production_plan/production_plan.js b/erpnext/manufacturing/doctype/production_plan/production_plan.js
index 71af0d8d290..e13dd716417 100644
--- a/erpnext/manufacturing/doctype/production_plan/production_plan.js
+++ b/erpnext/manufacturing/doctype/production_plan/production_plan.js
@@ -230,6 +230,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 {
diff --git a/erpnext/manufacturing/doctype/production_plan/production_plan.py b/erpnext/manufacturing/doctype/production_plan/production_plan.py
index 2401ec29308..07d091373cf 100644
--- a/erpnext/manufacturing/doctype/production_plan/production_plan.py
+++ b/erpnext/manufacturing/doctype/production_plan/production_plan.py
@@ -31,6 +31,9 @@ from pypika.terms import ExistsCriterion
 
 from erpnext.manufacturing.doctype.bom.bom import get_children as get_bom_children
 from erpnext.manufacturing.doctype.bom.bom import validate_bom_no
+from erpnext.manufacturing.doctype.production_plan.work_order_quantities import (
+	ProductionPlanWorkOrderQuantities,
+)
 from erpnext.manufacturing.doctype.work_order.work_order import get_item_details
 from erpnext.setup.doctype.item_group.item_group import get_item_group_defaults
 from erpnext.stock.doctype.item.item import get_uom_conv_factor
@@ -122,6 +125,12 @@ class ProductionPlan(Document):
 			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")
 
@@ -728,6 +737,7 @@ class ProductionPlan(Document):
 
 	def get_production_items(self):
 		item_dict = {}
+		pending = ProductionPlanWorkOrderQuantities(self.name).get_pending_quantities(self)
 
 		for d in self.po_items:
 			item_details = {
@@ -750,29 +760,12 @@ class ProductionPlan(Document):
 				"project": self.project,
 			}
 
-			key = (d.item_code, d.sales_order, d.sales_order_item, d.warehouse, d.planned_start_date)
-			if self.combine_items:
-				key = (d.item_code, d.sales_order, d.warehouse, d.planned_start_date)
-
-			if not d.sales_order:
-				key = (d.name, d.item_code, d.warehouse, d.planned_start_date)
-
 			if not item_details["project"] and d.sales_order:
 				item_details["project"] = frappe.get_cached_value("Sales Order", d.sales_order, "project")
 
-			if self.get_items_from == "Material Request":
-				item_details.update({"qty": d.planned_qty})
-				item_dict[
-					(d.item_code, d.material_request_item, d.warehouse, d.planned_start_date)
-				] = item_details
-			else:
-				item_details.update(
-					{
-						"qty": flt(item_dict.get(key, {}).get("qty"))
-						+ (flt(d.planned_qty) - flt(d.ordered_qty))
-					}
-				)
-				item_dict[key] = item_details
+			item_details["qty"] = pending["production_plan_item"][d.name]
+			# A Work Order can reference only one Production Plan row.
+			item_dict[d.name] = item_details
 
 		return item_dict
 
@@ -780,6 +773,7 @@ class ProductionPlan(Document):
 	def make_work_order(self):
 		from erpnext.manufacturing.doctype.work_order.work_order import get_default_warehouse
 
+		self.reload()
 		wo_list, po_list = [], []
 		subcontracted_po = {}
 		default_warehouses = get_default_warehouse(self.company)
@@ -809,6 +803,7 @@ class ProductionPlan(Document):
 				wo_list.append(work_order)
 
 	def make_work_order_for_subassembly_items(self, wo_list, subcontracted_po, default_warehouses):
+		pending = ProductionPlanWorkOrderQuantities(self.name).get_pending_quantities(self)
 		for row in self.sub_assembly_items:
 			if row.type_of_manufacturing == "Subcontract":
 				subcontracted_po.setdefault(row.supplier, []).append(row)
@@ -825,10 +820,9 @@ class ProductionPlan(Document):
 				"company": self.get("company"),
 			}
 
-			if flt(row.qty) <= flt(row.ordered_qty):
-				continue
-
-			self.prepare_data_for_sub_assembly_items(row, work_order_data)
+			self.prepare_data_for_sub_assembly_items(
+				row, work_order_data, pending["production_plan_sub_assembly_item"][row.name]
+			)
 
 			if work_order_data.get("qty") <= 0:
 				continue
@@ -837,7 +831,7 @@ class ProductionPlan(Document):
 			if work_order:
 				wo_list.append(work_order)
 
-	def prepare_data_for_sub_assembly_items(self, row, wo_data):
+	def prepare_data_for_sub_assembly_items(self, row, wo_data, pending_qty=None):
 		for field in [
 			"production_item",
 			"item_name",
@@ -853,7 +847,11 @@ class ProductionPlan(Document):
 			if row.get(field):
 				wo_data[field] = row.get(field)
 
-		wo_data["qty"] = flt(row.get("qty")) - flt(row.get("ordered_qty"))
+		if pending_qty is None:
+			pending_qty = ProductionPlanWorkOrderQuantities(self.name).get_pending_quantities(self)[
+				"production_plan_sub_assembly_item"
+			][row.name]
+		wo_data["qty"] = pending_qty
 
 		wo_data.update(
 			{
diff --git a/erpnext/manufacturing/doctype/production_plan/test_production_plan.py b/erpnext/manufacturing/doctype/production_plan/test_production_plan.py
index 1fb867daf3e..7cdf8c0ec03 100644
--- a/erpnext/manufacturing/doctype/production_plan/test_production_plan.py
+++ b/erpnext/manufacturing/doctype/production_plan/test_production_plan.py
@@ -1567,12 +1567,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():
@@ -1600,17 +1600,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])
@@ -1791,8 +1791,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()
diff --git a/erpnext/manufacturing/doctype/production_plan/test_work_order_quantities.py b/erpnext/manufacturing/doctype/production_plan/test_work_order_quantities.py
new file mode 100644
index 00000000000..b5062980de0
--- /dev/null
+++ b/erpnext/manufacturing/doctype/production_plan/test_work_order_quantities.py
@@ -0,0 +1,532 @@
+# 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.production_plan.work_order_quantities import (
+	ProductionPlanWorkOrderQuantities,
+)
+from erpnext.manufacturing.doctype.work_order.work_order import (
+	OverProductionError,
+	StockOverProductionError,
+	close_work_order,
+	stop_unstop,
+)
+from erpnext.manufacturing.doctype.work_order.work_order import make_stock_entry as make_se_from_wo
+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),
+		):
+			bom = make_bom(item=item, raw_materials=[material], do_not_save=True)
+			bom.process_loss_percentage = 10
+			bom.insert().submit()
+		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_cumulative_manufacture_loss_exceeds_work_order(self):
+		plan = self.make_plan()
+		for field in (None, *REFERENCE_FIELDS):
+			with self.subTest(field=field):
+				first = self.create_work_order(plan, field or "production_plan_item")
+				if field is None:
+					first.production_plan = None
+					first.production_plan_item = None
+				first.qty = 100
+				first.submit()
+				self.manufacture_with_loss(first, loss_qty=99)
+				second_entry = self.manufacture_with_loss(first, loss_qty=99, submit=False)
+				self.assert_manufacture_rejected(second_entry, StockOverProductionError)
+				self.assertEqual(first.reload().produced_qty, 1)
+				self.assertEqual(first.process_loss_qty, 99)
+				if field:
+					self.assert_pending_qty(plan, field, 99)
+					replacement = self.create_work_order(plan, field)
+					replacement.submit()
+					self.assert_pending_qty(plan, field, 0)
+
+	def test_cumulative_manufacture_loss_respects_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.submit()
+				self.manufacture_with_loss(first, qty=50)
+				self.manufacture_with_loss(first, qty=50)
+				last_entry = self.manufacture_with_loss(first, qty=10)
+				self.assertEqual(first.reload().produced_qty, 99)
+				self.assertEqual(first.process_loss_qty, 11)
+				self.assert_pending_qty(plan, field, 1)
+				excess = self.manufacture_with_loss(first, qty=1, submit=False)
+				self.assert_manufacture_rejected(excess, StockOverProductionError)
+				excess.delete()
+				last_entry.cancel()
+				self.assertEqual(first.reload().produced_qty, 90)
+				self.assertEqual(first.process_loss_qty, 10)
+				self.manufacture_with_loss(first, qty=10)
+
+	@ERPNextTestSuite.change_settings("System Settings", {"float_precision": 6})
+	def test_cumulative_fractional_manufacture_loss(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.submit()
+				self.manufacture_with_loss(first, qty=0.1, loss_qty=0.025)
+				self.manufacture_with_loss(first, qty=0.2, loss_qty=0.05)
+				self.assertAlmostEqual(first.reload().produced_qty, 0.225)
+				self.assertAlmostEqual(first.process_loss_qty, 0.075)
+				excess = self.manufacture_with_loss(first, qty=0.001, submit=False)
+				self.assert_manufacture_rejected(excess, StockOverProductionError)
+
+	def test_existing_excess_loss_preserves_produced_quantity(self):
+		for field in REFERENCE_FIELDS:
+			for loss_qty, produced_qty in ((198, 2), (100, 2), (20, 90), (198, 0)):
+				with self.subTest(field=field, loss_qty=loss_qty, produced_qty=produced_qty):
+					plan = self.make_plan()
+					first = self.create_work_order(plan, field)
+					first.submit()
+					# Reproduce records saved before cumulative manufacture validation existed.
+					first.db_set({"process_loss_qty": loss_qty, "produced_qty": produced_qty})
+					pending_qty = 100 - produced_qty
+					self.assert_pending_qty(plan, field, pending_qty)
+					replacement = self.create_work_order(plan, field)
+					self.assertEqual(replacement.qty, pending_qty)
+					self.assert_overproduction(replacement, pending_qty + 1)
+					replacement.qty = pending_qty
+					replacement.submit()
+					self.assert_pending_qty(plan, field, 0)
+					quantities = ProductionPlanWorkOrderQuantities(plan.name)
+					quantities.validate_work_order(first, process_loss_qty=loss_qty)
+					with self.assertRaises(OverProductionError):
+						quantities.validate_work_order(first, process_loss_qty=0)
+
+	def test_more_production_cannot_consume_replacement_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.submit()
+				self.manufacture_with_loss(first, loss_qty=99)
+				replacement = self.create_work_order(plan, field)
+				replacement.qty = 109
+				replacement.submit()
+				# This fits the first Work Order's allowance, but exceeds the plan's 110 units.
+				excess = self.manufacture_with_loss(first, qty=10, loss_qty=9, submit=False)
+				self.assert_manufacture_rejected(excess, OverProductionError)
+				self.assertEqual(first.reload().produced_qty, 1)
+				self.assertEqual(first.process_loss_qty, 99)
+
+	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, *, loss_qty=None, submit=True):
+		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))
+		if loss_qty is not None:
+			entry.process_loss_qty = loss_qty
+			entry.process_loss_percentage = loss_qty / entry.fg_completed_qty * 100
+			for item in entry.items:
+				if item.is_finished_item:
+					item.qty = entry.fg_completed_qty - loss_qty
+		if submit:
+			entry.submit()
+		else:
+			entry.save()
+		return entry
+
+	def assert_manufacture_rejected(self, entry, exception):
+		frappe.db.savepoint("excess_manufacture")
+		try:
+			with self.assertRaises(exception):
+				entry.submit()
+		finally:
+			frappe.db.rollback(save_point="excess_manufacture")
+		self.assertEqual(entry.reload().docstatus, 0)
+
+	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]
diff --git a/erpnext/manufacturing/doctype/production_plan/work_order_quantities.py b/erpnext/manufacturing/doctype/production_plan/work_order_quantities.py
new file mode 100644
index 00000000000..6d6d0a800c9
--- /dev/null
+++ b/erpnext/manufacturing/doctype/production_plan/work_order_quantities.py
@@ -0,0 +1,143 @@
+# 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, precision)
+		committed_qty = flt(self._get_committed_qty(work_order, process_loss_qty), precision)
+		if committed_qty > maximum_qty:
+			frappe.throw(
+				_(
+					"Row {0} in {1} {2}: Work Order quantity after process loss {3} exceeds the remaining allowed quantity {4}."
+				).format(
+					row.idx,
+					_(row.doctype),
+					get_link_to_form("Production Plan", self.production_plan),
+					committed_qty,
+					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
+	):
+		table = frappe.qb.DocType("Work Order")
+		query = (
+			frappe.qb.from_(table)
+			.select(
+				table.production_plan_item,
+				table.production_plan_sub_assembly_item,
+				table.qty,
+				table.produced_qty,
+				table.process_loss_qty,
+			)
+			.where((table.production_plan == self.production_plan) & (table.docstatus == 1))
+			.orderby(table.name)
+		)
+		if exclude_work_order:
+			query = query.where(table.name != exclude_work_order)
+		if reference_field:
+			query = query.where(table[reference_field] == reference_name)
+		# Use a current locking read so concurrent submissions see committed quantities.
+		if for_update:
+			query = query.for_update()
+		work_orders = query.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]] += self._get_committed_qty(
+					work_order, work_order.process_loss_qty
+				)
+		return quantities
+
+	def _get_committed_qty(self, work_order, process_loss_qty):
+		# Excess loss in existing records must not erase finished goods already produced.
+		return max(0, flt(work_order.produced_qty), flt(work_order.qty) - flt(process_loss_qty))
diff --git a/erpnext/manufacturing/doctype/work_order/work_order.py b/erpnext/manufacturing/doctype/work_order/work_order.py
index 68a71e69569..9472aaf71d8 100644
--- a/erpnext/manufacturing/doctype/work_order/work_order.py
+++ b/erpnext/manufacturing/doctype/work_order/work_order.py
@@ -34,6 +34,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.work_order_quantities import (
+	ProductionPlanWorkOrderQuantities,
+)
 from erpnext.setup.doctype.item_group.item_group import get_item_group_defaults
 from erpnext.stock.doctype.batch.batch import make_batch
 from erpnext.stock.doctype.item.item import get_item_defaults, validate_end_of_life
@@ -799,6 +802,8 @@ class WorkOrder(Document):
 		if self.track_semi_finished_goods:
 			return
 
+		# Lock the plan row before any Work Order quantity update takes a row lock.
+		self.set_process_loss_qty()
 		allowance_percentage = flt(
 			frappe.db.get_single_value("Manufacturing Settings", "overproduction_percentage_for_work_order")
 		)
@@ -825,16 +830,26 @@ class WorkOrder(Document):
 				)
 
 			completed_qty = self.qty + (allowance_percentage / 100 * self.qty)
-			if qty > completed_qty:
+			qty_to_validate = qty + flt(self.process_loss_qty) if purpose == "Manufacture" else qty
+			precision = self.precision(fieldname)
+			if flt(qty_to_validate, precision) > flt(completed_qty, precision):
 				frappe.throw(
 					_("{0} ({1}) cannot be greater than planned quantity ({2}) in Work Order {3}").format(
-						_(self.meta.get_label(fieldname)), qty, completed_qty, self.name
+						_("Manufactured Qty (including Process Loss)")
+						if purpose == "Manufacture"
+						else _(self.meta.get_label(fieldname)),
+						flt(qty_to_validate, precision),
+						completed_qty,
+						self.name,
 					),
 					StockOverProductionError,
 				)
 
 			self.db_set(fieldname, qty)
-			self.set_process_loss_qty()
+			if purpose == "Manufacture" and self.production_plan:
+				ProductionPlanWorkOrderQuantities(self.production_plan).validate_work_order(
+					self, process_loss_qty=self.process_loss_qty
+				)
 
 			from erpnext.selling.doctype.sales_order.sales_order import update_produced_qty_in_so_item
 
@@ -905,7 +920,20 @@ class WorkOrder(Document):
 		return flt(query.run()[0][0])
 
 	def set_process_loss_qty(self):
-		self.db_set("process_loss_qty", self._process_loss_qty())
+		quantities = None
+		if self.docstatus == 1 and self.production_plan:
+			quantities = ProductionPlanWorkOrderQuantities(self.production_plan)
+			quantities.lock_plan_row(self)
+
+		process_loss_qty = self._process_loss_qty()
+		if quantities:
+			previous_loss_qty = frappe.db.get_value(
+				"Work Order", self.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, process_loss_qty=process_loss_qty)
+		self.db_set("process_loss_qty", process_loss_qty)
 
 	def _process_loss_qty(self):
 		if self.track_semi_finished_goods:
@@ -950,6 +978,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):
@@ -1622,36 +1652,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(
 				_(
diff --git a/erpnext/stock/doctype/stock_entry/stock_entry.py b/erpnext/stock/doctype/stock_entry/stock_entry.py
index 5e97918137c..8a3a9e63ac9 100644
--- a/erpnext/stock/doctype/stock_entry/stock_entry.py
+++ b/erpnext/stock/doctype/stock_entry/stock_entry.py
@@ -33,6 +33,9 @@ from erpnext.manufacturing.doctype.bom.bom import (
 	get_secondary_items_from_sub_assemblies,
 	validate_bom_no,
 )
+from erpnext.manufacturing.doctype.production_plan.work_order_quantities import (
+	ProductionPlanWorkOrderQuantities,
+)
 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.doctype.batch.batch import get_batch_qty
@@ -2517,6 +2520,8 @@ class StockEntry(StockController, SubcontractingInwardController):
 		if self.work_order:
 			pro_doc = frappe.get_doc("Work Order", self.work_order)
 			_validate_work_order(pro_doc)
+			if pro_doc.production_plan:
+				ProductionPlanWorkOrderQuantities(pro_doc.production_plan).lock_plan_row(pro_doc)
 
 			if self.fg_completed_qty:
 				if self.docstatus == 1: