feat: validate stock value and stock closing entry before period closing (#57811)

* feat: validate stock value and stock closing entry before period closing

* fix: do not accept scoped stock closing entries as period closing prerequisite

* feat: seed batch valuation from stock closing balance and freeze closed-period stock

.github/POSTGRES_COMPATIBILITY.md

@@ -170,6 +170,13 @@ audit of these fixes found four recurring mistakes:
 - **Fabricated arithmetic** — `Sum(x) * Max(y)` where `y` varies within the group invents a
   number no row ever had (and `Max` biases it upward) — poisonous when it feeds validation,
   budgets, valuation, or GL/stock values. Fix per-row: `Sum(x * y)`.
+- **Collation-dependent pick (text columns)** — `Max()`/`Min()` over text is a *sort*, and the two
+  engines sort text differently: MariaDB's `utf8mb4` collations fold case, PostgreSQL (as CI runs
+  it) orders by byte value. `MAX('abc', 'ABD')` is `ABD` on MariaDB and `abc` on PostgreSQL. So a
+  `Max()` over a text column that varies **in case** within its group is a live P2 divergence, not
+  the arbitrary-pick preservation the wrap is usually justified as. Confirmed on CI; see #56241.
+  Note a local macOS PostgreSQL gives a **false all-clear** — its collation happens to agree with
+  MariaDB on case. Fix: take a representative row rather than sorting text.
 - **Wrong bound** — where the value has a semantic, pick the bound deliberately:
   `Min(schedule_date)` for a "required by", `Min(idx)` for first-line ordering, a qty-weighted
   average for a rate. A blind `Max` can understate urgency or overstate a figure.

.github/workflows/patch.yml

@@ -171,7 +171,30 @@ jobs:
           update_to_version 16 3.14
 
           echo "Updating to latest version"
-          git -C "apps/frappe" fetch --depth 1 upstream "${GITHUB_BASE_REF:-${GITHUB_REF##*/}}"
+          fallback_to_develop=0
+          if [ -n "${GITHUB_BASE_REF:-}" ]; then
+            frappe_ref="refs/heads/$GITHUB_BASE_REF"
+            fallback_to_develop=1
+          elif [ "${GITHUB_REF_TYPE:-}" = "branch" ]; then
+            frappe_ref="$GITHUB_REF"
+            fallback_to_develop=1
+          elif [ "${GITHUB_REF_TYPE:-}" = "tag" ]; then
+            frappe_ref="$GITHUB_REF"
+          else
+            echo "Unsupported GitHub ref type: '${GITHUB_REF_TYPE:-unset}'"
+            exit 1
+          fi
+
+          ls_remote_status=0
+          git -C "apps/frappe" ls-remote --exit-code upstream "$frappe_ref" >/dev/null \
+            || ls_remote_status=$?
+          if [ "$ls_remote_status" -eq 2 ] && [ "$fallback_to_develop" -eq 1 ]; then
+            echo "frappe has no '$frappe_ref'; falling back to develop"
+            frappe_ref=refs/heads/develop
+          elif [ "$ls_remote_status" -ne 0 ]; then
+            exit "$ls_remote_status"
+          fi
+          git -C "apps/frappe" fetch --depth 1 upstream "$frappe_ref"
           git -C "apps/frappe" checkout -q -f FETCH_HEAD
           git -C "apps/erpnext" checkout -q -f "$GITHUB_SHA"
 

.github/workflows/review-translation-changes.yaml

@@ -23,3 +23,5 @@ jobs:
 
     steps:
       - uses: alyf-de/po-review-action@v1.1.0
+        with:
+          hidden-po-files: eo.po

CODEOWNERS

@@ -7,6 +7,7 @@ erpnext/accounts/               @ruthra-kumar
 erpnext/assets/                 @khushi8112
 erpnext/regional                @ruthra-kumar
 erpnext/selling                 @ruthra-kumar
+banking/                        @nikkothari22
 
 erpnext/buying/                 @rohitwaghchaure @mihir-kandoi
 erpnext/maintenance/            @rohitwaghchaure @mihir-kandoi

banking/src/components/features/BankReconciliation/CompanySelector.tsx

@@ -19,13 +19,22 @@ import {
 import { cn } from "@/lib/utils"
 import _ from "@/lib/translate"
 import { selectedBankAccountAtom } from "./bankRecAtoms"
+import { useFrappeGetDocList } from "frappe-react-sdk"
+import ErrorBanner from "@/components/ui/error-banner"
 
 const CompanySelector = ({ onChange }: { onChange?: (company: string) => void }) => {
     const [open, setOpen] = useState(false)
     const [searchQuery, setSearchQuery] = useState("")
 
-    // eslint-disable-next-line @typescript-eslint/no-explicit-any
-    const options = window.frappe?.boot?.docs?.filter((doc: Record<string, any>) => doc.doctype === ":Company").map((company: Record<string, any>) => company.name) || []
+    const { data: companies, error } = useFrappeGetDocList("Company", {
+        limit: 0,
+        fields: ["name"],
+    }, 'company_list', {
+        revalidateOnFocus: false,
+        revalidateOnReconnect: false,
+    })
+
+    const options = companies?.map((company: { name: string }) => company.name) || []
 
     const setSelectedCompany = useSetAtom(selectedCompanyAtom)
     const setSelectedBankAccount = useSetAtom(selectedBankAccountAtom)
@@ -42,6 +51,10 @@ const CompanySelector = ({ onChange }: { onChange?: (company: string) => void })
         }
     }
 
+    if (error) {
+        return <ErrorBanner error={error} />
+    }
+
     return (<Popover open={open} onOpenChange={setOpen}>
         <PopoverTrigger asChild>
             <Button

banking/src/hooks/useCurrentCompany.ts

@@ -1,7 +1,9 @@
 import { useAtomValue } from "jotai"
 import { atomWithStorage } from "jotai/utils"
 
-export const selectedCompanyAtom = atomWithStorage<string>('bank-rec-selected-company', window.frappe?.boot?.user?.defaults?.company || '')
+export const selectedCompanyAtom = atomWithStorage<string>('bank-rec-selected-company', window.frappe?.boot?.user?.defaults?.company || '', undefined, {
+    getOnInit: true,
+})
 
 export const useCurrentCompany = () => {
     const selectedCompany = useAtomValue(selectedCompanyAtom)

erpnext/accounts/doctype/dunning/dunning.py

@@ -275,6 +275,7 @@ def get_linked_dunnings_as_per_state(sales_invoice, state):
 		.join(overdue_payment)
 		.on(overdue_payment.parent == dunning.name)
 		.select(dunning.name)
+		.distinct()
 		.where(
 			(dunning.status == state)
 			& (dunning.docstatus != 2)

erpnext/accounts/doctype/dunning/test_dunning.py

@@ -123,6 +123,41 @@ class TestDunning(ERPNextTestSuite):
 		self.assertEqual(sales_invoice.status, "Overdue")
 		self.assertEqual(dunning.status, "Unresolved")
 
+	def test_payment_against_invoice_with_multiple_overdue_installments_in_dunning(self):
+		"""
+		When an invoice has more than one overdue installment, its Dunning holds one
+		Overdue Payment row per installment. Submitting a Payment Entry for the invoice
+		must resolve the Dunning without raising a TimestampMismatchError caused by the
+		same Dunning being loaded and saved more than once.
+		"""
+		create_payment_terms_template_for_dunning()
+		# Post far enough in the past that BOTH installments (5 and 10 credit days) are overdue.
+		sales_invoice = create_sales_invoice_against_cost_center(
+			posting_date=add_days(today(), -15),
+			qty=1,
+			rate=100,
+			do_not_submit=True,
+		)
+		sales_invoice.payment_terms_template = "_Test 50-50 for Dunning"
+		sales_invoice.submit()
+
+		dunning = create_dunning_from_sales_invoice(sales_invoice.name)
+		# Two overdue installments -> two overdue payment rows for the same invoice.
+		self.assertEqual(len(dunning.overdue_payments), 2)
+		dunning.submit()
+		self.assertEqual(dunning.status, "Unresolved")
+
+		# Pay the invoice in full. This previously raised TimestampMismatchError on the Dunning.
+		pe = get_payment_entry("Sales Invoice", sales_invoice.name)
+		pe.reference_no, pe.reference_date = "3", nowdate()
+		pe.insert()
+		pe.submit()
+
+		sales_invoice.reload()
+		dunning.reload()
+		self.assertEqual(sales_invoice.outstanding_amount, 0)
+		self.assertEqual(dunning.status, "Resolved")
+
 	def test_dunning_resolution_from_credit_note(self):
 		"""
 		Test that dunning is resolved when a credit note is issued against the original invoice.

erpnext/accounts/doctype/journal_entry/journal_entry.js

@@ -235,7 +235,7 @@ Object.assign(erpnext.journal_entry, {
 	lock_reversal_entry(frm) {
 		frm.fields
 			.filter((field) => field.has_input)
-			.filter((field) => field.df.fieldname != "posting_date")
+			.filter((field) => !["posting_date", "custom_remark", "remark"].includes(field.df.fieldname))
 			.forEach((field) => frm.set_df_property(field.df.fieldname, "read_only", 1));
 		frm.set_df_property("accounts", "read_only", 1);
 	},

erpnext/accounts/doctype/payment_reconciliation/test_payment_reconciliation.py

@@ -187,6 +187,103 @@ class TestPaymentReconciliation(ERPNextTestSuite):
 		)
 		return je
 
+	def test_voucher_outstanding_metadata_comes_from_one_ledger_entry(self):
+		"""cost_center and remarks must describe the same Payment Ledger Entry.
+
+		A voucher can post several ledger entries for one party with different cost centers and
+		remarks. Aggregating each column on its own can pair one entry's cost center with another's
+		remarks -- a row that was never posted -- and because Max() over text is a sort, MariaDB and
+		PostgreSQL can pick differently on top of that.
+		"""
+		from erpnext.accounts.utils import QueryPaymentLedger
+
+		je = frappe.new_doc("Journal Entry")
+		je.posting_date = nowdate()
+		je.company = self.company
+		je.user_remark = "aaa base remark"
+		for cost_center, remark, amount in (
+			(self.main_cc, "aaa main line", 100),
+			(self.sub_cc, "zzz sub line", 50),
+		):
+			je.append(
+				"accounts",
+				{
+					"account": self.debit_to,
+					"party_type": "Customer",
+					"party": self.customer,
+					"cost_center": cost_center,
+					"user_remark": remark,
+					"debit_in_account_currency": amount,
+				},
+			)
+		je.append(
+			"accounts", {"account": self.cash, "cost_center": self.main_cc, "credit_in_account_currency": 150}
+		)
+		je.save()
+		je.submit()
+
+		posted = {
+			(row.cost_center, row.remarks)
+			for row in frappe.get_all(
+				"Payment Ledger Entry",
+				filters={"voucher_no": je.name, "delinked": 0},
+				fields=["cost_center", "remarks"],
+			)
+		}
+		self.assertGreater(len(posted), 1, "fixture must post more than one ledger entry to be meaningful")
+
+		ledger = QueryPaymentLedger()
+		rows = ledger.get_voucher_outstandings(
+			vouchers=[frappe._dict(voucher_type="Journal Entry", voucher_no=je.name)]
+		)
+		self.assertTrue(rows)
+
+		for row in rows:
+			self.assertIn((row.cost_center, row.remarks), posted)
+
+	def test_voucher_outstanding_splits_by_party_account(self):
+		"""A voucher posting to two party accounts must report each account separately.
+
+		account is the join key between the amount and outstanding CTEs. Selecting Max(account)
+		while grouping without it made that key an aggregate over two different row sets, so the two
+		sides could pick different accounts, the join would miss and the outstanding come back NULL.
+		It also summed amounts across accounts that need not share a currency.
+		"""
+		from erpnext.accounts.utils import QueryPaymentLedger
+
+		second_receivable = "_Test Receivable - _TC"
+		je = frappe.new_doc("Journal Entry")
+		je.posting_date = nowdate()
+		je.company = self.company
+		je.user_remark = "two receivable accounts"
+		for account, amount in ((self.debit_to, 100), (second_receivable, 60)):
+			je.append(
+				"accounts",
+				{
+					"account": account,
+					"party_type": "Customer",
+					"party": self.customer,
+					"cost_center": self.main_cc,
+					"debit_in_account_currency": amount,
+				},
+			)
+		je.append(
+			"accounts", {"account": self.cash, "cost_center": self.main_cc, "credit_in_account_currency": 160}
+		)
+		je.save()
+		je.submit()
+
+		rows = QueryPaymentLedger().get_voucher_outstandings(
+			vouchers=[frappe._dict(voucher_type="Journal Entry", voucher_no=je.name)]
+		)
+		by_account = {row.account: row for row in rows}
+
+		self.assertEqual(set(by_account), {self.debit_to, second_receivable})
+		self.assertEqual(flt(by_account[self.debit_to].invoice_amount), 100)
+		self.assertEqual(flt(by_account[second_receivable].invoice_amount), 60)
+		for row in rows:
+			self.assertIsNotNone(row.outstanding)
+
 	def test_filter_min_max(self):
 		# check filter condition minimum and maximum amount
 		self.create_sales_invoice(qty=1, rate=300)
@@ -2402,6 +2499,86 @@ class TestPaymentReconciliation(ERPNextTestSuite):
 		self.assertEqual(flt(pr.allocation[0].get("difference_amount")), -5000.0)
 		pr.reconcile()
 
+	def test_foreign_currency_reverse_payment_entry_gain_for_supplier(self):
+		transaction_date = nowdate()
+		self.supplier = "_Test Supplier USD"
+		amount = 100
+		department = frappe.db.get_value("Department", {"company": self.company, "is_group": 0}, "name")
+
+		# Pay USD 100 at an exchange rate of 90.
+		pe = self.create_payment_entry(amount=amount, posting_date=transaction_date)
+		pe.payment_type = "Pay"
+		pe.party_type = "Supplier"
+		pe.party = self.supplier
+		pe.paid_from = self.cash
+		pe.paid_from_account_currency = "INR"
+		pe.target_exchange_rate = 90
+		pe.paid_amount = 90 * amount
+		pe.received_amount = amount
+		pe.paid_to = self.creditors_usd
+		pe.paid_to_account_currency = "USD"
+		pe.department = department
+		pe = pe.save().submit()
+
+		# Receive USD 100 from the supplier at an exchange rate of 100.
+		reverse_pe = self.create_payment_entry(amount=amount, posting_date=transaction_date)
+		reverse_pe.payment_type = "Receive"
+		reverse_pe.party_type = "Supplier"
+		reverse_pe.party = self.supplier
+		reverse_pe.paid_from = self.creditors_usd
+		reverse_pe.paid_from_account_currency = "USD"
+		reverse_pe.source_exchange_rate = 100
+		reverse_pe.paid_amount = amount
+		reverse_pe.received_amount = 100 * amount
+		reverse_pe.paid_to = self.cash
+		reverse_pe.paid_to_account_currency = "INR"
+		reverse_pe.department = department
+		reverse_pe = reverse_pe.save().submit()
+
+		pr = self.create_payment_reconciliation(party_is_customer=False)
+		pr.party = self.supplier
+		pr.receivable_payable_account = self.creditors_usd
+		pr.get_unreconciled_entries()
+		invoices = [invoice.as_dict() for invoice in pr.invoices]
+		payments = [payment.as_dict() for payment in pr.payments]
+		pr.allocate_entries(frappe._dict({"invoices": invoices, "payments": payments}))
+		for row in pr.allocation:
+			row.department = department
+
+		self.assertEqual(flt(pr.allocation[0].difference_amount), 1000)
+		pr.reconcile()
+
+		gain_loss_journal = frappe.db.get_value(
+			"Journal Entry Account",
+			{
+				"reference_type": reverse_pe.doctype,
+				"reference_name": reverse_pe.name,
+				"party": self.supplier,
+				"docstatus": 1,
+			},
+			"parent",
+		)
+		party_row = frappe.db.get_value(
+			"Journal Entry Account",
+			{"parent": gain_loss_journal, "party": self.supplier},
+			["debit", "credit"],
+			as_dict=True,
+		)
+		self.assertEqual(flt(party_row.debit), 1000)
+		self.assertEqual(flt(party_row.credit), 0)
+
+		party_gl_entries = frappe.get_all(
+			"GL Entry",
+			filters={
+				"voucher_no": ["in", [pe.name, reverse_pe.name, gain_loss_journal]],
+				"account": self.creditors_usd,
+				"party": self.supplier,
+				"is_cancelled": 0,
+			},
+			fields=["debit", "credit"],
+		)
+		self.assertEqual(flt(sum(row.debit - row.credit for row in party_gl_entries)), 0)
+
 	def test_foreign_currency_reverse_journal_entry_against_journal_entry_for_customer(self):
 		transaction_date = nowdate()
 		customer = self.customer_usd

erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher.py

@@ -6,8 +6,10 @@ import copy
 
 import frappe
 from frappe import _
-from frappe.utils import add_days, flt, formatdate, getdate
+from frappe.query_builder.functions import Max, Sum
+from frappe.utils import add_days, flt, fmt_money, formatdate, get_link_to_form, getdate
 
+from erpnext import is_perpetual_inventory_enabled
 from erpnext.accounts.doctype.account_closing_balance.account_closing_balance import (
 	make_closing_entries,
 )
@@ -17,6 +19,8 @@ from erpnext.accounts.doctype.accounting_dimension.accounting_dimension import (
 from erpnext.accounts.general_ledger import check_freezing_date, is_immutable_ledger_enabled
 from erpnext.accounts.utils import get_account_currency, get_fiscal_year
 from erpnext.controllers.accounts_controller import AccountsController
+from erpnext.stock.doctype.stock_closing_entry.stock_closing_entry import apply_unscoped_filters
+from erpnext.stock.utils import get_stock_value_on
 
 
 class PeriodClosingVoucher(AccountsController):
@@ -141,6 +145,121 @@ class PeriodClosingVoucher(AccountsController):
 		if account_currency != company_currency:
 			frappe.throw(_("Currency of the Closing Account must be {0}").format(company_currency))
 
+	def before_submit(self):
+		if not self.has_stock_transactions():
+			return
+
+		self.validate_stock_accounts_balance()
+		self.validate_stock_closing_entry()
+
+	def has_stock_transactions(self):
+		if not is_perpetual_inventory_enabled(self.company):
+			return False
+
+		return bool(
+			frappe.db.exists(
+				"Stock Ledger Entry",
+				{
+					"company": self.company,
+					"is_cancelled": 0,
+					"posting_date": ("<=", self.period_end_date),
+				},
+			)
+		)
+
+	def validate_stock_accounts_balance(self):
+		precision = frappe.get_precision("GL Entry", "debit")
+		account_balance = flt(self.get_stock_accounts_balance(), precision)
+		stock_value = flt(
+			get_stock_value_on(posting_date=self.period_end_date, company=self.company), precision
+		)
+
+		if account_balance == stock_value:
+			return
+
+		currency = frappe.get_cached_value("Company", self.company, "default_currency")
+		frappe.throw(
+			_(
+				"The closing balance {0} of the Stock Asset accounts does not match the closing value {1} of the Stock Balance report as on {2}. Resolve the difference using the Stock Ledger Variance report before closing the period."
+			).format(
+				frappe.bold(fmt_money(account_balance, currency=currency)),
+				frappe.bold(fmt_money(stock_value, currency=currency)),
+				frappe.bold(formatdate(self.period_end_date)),
+			),
+			title=_("Stock Value Mismatch"),
+		)
+
+	def get_stock_accounts_balance(self):
+		gle = frappe.qb.DocType("GL Entry")
+		account = frappe.qb.DocType("Account")
+
+		stock_accounts = (
+			frappe.qb.from_(account)
+			.select(account.name)
+			.where(
+				(account.account_type == "Stock")
+				& (account.company == self.company)
+				& (account.is_group == 0)
+			)
+		)
+
+		balance = (
+			frappe.qb.from_(gle)
+			.select(Sum(gle.debit - gle.credit))
+			.where(
+				(gle.company == self.company)
+				& (gle.is_cancelled == 0)
+				& (gle.posting_date <= self.period_end_date)
+				& gle.account.isin(stock_accounts)
+			)
+		).run()
+
+		return flt(balance[0][0]) if balance else 0.0
+
+	def validate_stock_closing_entry(self):
+		closing_entry = frappe.db.get_value(
+			"Stock Closing Entry",
+			apply_unscoped_filters(
+				{"company": self.company, "to_date": self.period_end_date, "docstatus": 1}
+			),
+			["name", "status", "modified"],
+			as_dict=True,
+		)
+
+		if not closing_entry:
+			frappe.throw(
+				_(
+					"Create a Stock Closing Entry for the entire company with To Date as {0} before submitting the Period Closing Voucher."
+				).format(frappe.bold(formatdate(self.period_end_date))),
+				title=_("Stock Closing Entry Required"),
+			)
+
+		if closing_entry.status != "Completed":
+			frappe.throw(
+				_(
+					"The Stock Closing Entry for {0} is not completed yet. Wait for it to complete before submitting the Period Closing Voucher."
+				).format(frappe.bold(formatdate(self.period_end_date))),
+				title=_("Stock Closing Entry In Progress"),
+			)
+
+		self.validate_stock_closing_entry_is_fresh(closing_entry)
+
+	def validate_stock_closing_entry_is_fresh(self, closing_entry):
+		sle = frappe.qb.DocType("Stock Ledger Entry")
+		last_change = (
+			frappe.qb.from_(sle)
+			.select(Max(sle.modified))
+			.where((sle.company == self.company) & (sle.posting_date <= self.period_end_date))
+		).run()
+
+		if last_change and last_change[0][0] and last_change[0][0] > closing_entry.modified:
+			frappe.throw(
+				_(
+					"Stock transactions were created or modified after the Stock Closing Entry {0} was generated. Regenerate it before submitting the Period Closing Voucher."
+				).format(get_link_to_form("Stock Closing Entry", closing_entry.name)),
+				title=_("Stock Closing Entry Outdated"),
+			)
+
 	def on_submit(self):
 		self.db_set("gle_processing_status", "In Progress")
 		if frappe.get_single_value("Accounts Settings", "use_legacy_controller_for_pcv"):

erpnext/accounts/doctype/period_closing_voucher/test_period_closing_voucher.py

@@ -2,7 +2,7 @@
 # License: GNU General Public License v3. See license.txt
 
 import frappe
-from frappe.utils import today
+from frappe.utils import flt, today
 
 from erpnext.accounts.doctype.finance_book.test_finance_book import create_finance_book
 from erpnext.accounts.doctype.journal_entry.test_journal_entry import make_journal_entry
@@ -386,6 +386,218 @@ class TestPeriodClosingVoucher(ERPNextTestSuite):
 		self.assertEqual(acb_figures["Cash"][key_for(cc1)], 400)
 		self.assertEqual(acb_figures["Cash"][key_for(cc2)], 200)
 
+	def test_stock_validations_before_period_closing(self):
+		from unittest.mock import patch
+
+		from frappe.custom.doctype.custom_field.custom_field import create_custom_fields
+
+		from erpnext.stock.doctype.item.test_item import make_item
+		from erpnext.stock.doctype.stock_entry.stock_entry_utils import make_stock_entry
+
+		create_custom_fields(
+			{
+				"Stock Closing Entry": [
+					{
+						"fieldname": "warehouse",
+						"label": "Warehouse",
+						"fieldtype": "Link",
+						"options": "Warehouse",
+					}
+				]
+			}
+		)
+
+		item = make_item("Test PCV Stock Item", {"is_stock_item": 1})
+		se = make_stock_entry(
+			item_code=item.name,
+			qty=10,
+			rate=100,
+			to_warehouse="Stores - TPC",
+			company="Test PCV Company",
+			posting_date="2021-03-15",
+		)
+
+		pcv = self.make_period_closing_voucher(posting_date="2021-03-31", submit=False)
+		self.assertRaisesRegex(frappe.ValidationError, "Create a Stock Closing Entry", pcv.submit)
+
+		sce = frappe.get_doc(
+			{
+				"doctype": "Stock Closing Entry",
+				"company": "Test PCV Company",
+				"from_date": pcv.period_start_date,
+				"to_date": pcv.period_end_date,
+				"warehouse": "Stores - TPC",
+			}
+		).insert()
+
+		with patch("erpnext.stock.doctype.stock_closing_entry.stock_closing_entry.enqueue"):
+			sce.submit()
+
+		sce.db_set("status", "Completed")
+
+		pcv.reload()
+		self.assertRaisesRegex(frappe.ValidationError, "Create a Stock Closing Entry", pcv.submit)
+
+		frappe.db.set_value("Stock Closing Entry", sce.name, {"warehouse": None, "status": "In Progress"})
+
+		pcv.reload()
+		self.assertRaisesRegex(frappe.ValidationError, "is not completed yet", pcv.submit)
+
+		sce.create_stock_closing_balance_entries()
+		sce.db_set("status", "Completed")
+
+		sle = frappe.db.get_value(
+			"Stock Ledger Entry",
+			{"voucher_no": se.name},
+			["name", "stock_value_difference"],
+			as_dict=1,
+		)
+		frappe.db.set_value(
+			"Stock Ledger Entry", sle.name, "stock_value_difference", sle.stock_value_difference + 100
+		)
+
+		pcv.reload()
+		self.assertRaisesRegex(frappe.ValidationError, "does not match", pcv.submit)
+
+		frappe.db.set_value(
+			"Stock Ledger Entry", sle.name, "stock_value_difference", sle.stock_value_difference
+		)
+
+		pcv.reload()
+		self.assertRaisesRegex(frappe.ValidationError, "Regenerate", pcv.submit)
+
+		self.rebuild_stock_closing_balance(sce)
+		pcv.reload()
+		pcv.submit()
+		self.assertEqual(pcv.docstatus, 1)
+
+	def test_batch_valuation_seeded_from_stock_closing_after_period_closing(self):
+		from erpnext.stock.doctype.item.test_item import make_item
+		from erpnext.stock.doctype.serial_and_batch_bundle.test_serial_and_batch_bundle import (
+			get_batch_from_bundle,
+		)
+		from erpnext.stock.doctype.stock_entry.stock_entry_utils import make_stock_entry
+
+		item = make_item(
+			"Test PCV Batch Item",
+			{
+				"is_stock_item": 1,
+				"has_batch_no": 1,
+				"create_new_batch": 1,
+				"batch_number_series": "TPCVB.####",
+			},
+		)
+		se1 = make_stock_entry(
+			item_code=item.name,
+			qty=10,
+			rate=100,
+			to_warehouse="Stores - TPC",
+			company="Test PCV Company",
+			posting_date="2021-03-15",
+		)
+		batch_no = get_batch_from_bundle(se1.items[0].serial_and_batch_bundle)
+		make_stock_entry(
+			item_code=item.name,
+			qty=10,
+			rate=200,
+			to_warehouse="Stores - TPC",
+			company="Test PCV Company",
+			posting_date="2021-06-15",
+			batch_no=batch_no,
+		)
+
+		pcv = self.make_period_closing_voucher(posting_date="2021-03-31", submit=False)
+		sce = self.make_completed_stock_closing_entry(pcv.period_start_date, pcv.period_end_date)
+		pcv.reload()
+		pcv.submit()
+
+		outward = make_stock_entry(
+			item_code=item.name,
+			qty=5,
+			from_warehouse="Stores - TPC",
+			company="Test PCV Company",
+			posting_date="2022-04-01",
+			batch_no=batch_no,
+		)
+		stock_value_difference = frappe.db.get_value(
+			"Stock Ledger Entry",
+			{"voucher_no": outward.name, "is_cancelled": 0},
+			"stock_value_difference",
+		)
+		self.assertEqual(flt(stock_value_difference, 2), -750.0)
+
+		self.assertRaisesRegex(
+			frappe.ValidationError,
+			"frozen",
+			make_stock_entry,
+			item_code=item.name,
+			qty=1,
+			rate=100,
+			to_warehouse="Stores - TPC",
+			company="Test PCV Company",
+			posting_date="2021-05-01",
+		)
+		self.assertRaisesRegex(frappe.ValidationError, "frozen", se1.cancel)
+		self.assertRaisesRegex(frappe.ValidationError, "closed accounting period", sce.cancel)
+
+	def test_period_closing_blocks_stale_stock_closing_entry(self):
+		from erpnext.stock.doctype.item.test_item import make_item
+		from erpnext.stock.doctype.stock_entry.stock_entry_utils import make_stock_entry
+
+		item = make_item("Test PCV Stock Item", {"is_stock_item": 1})
+		make_stock_entry(
+			item_code=item.name,
+			qty=10,
+			rate=100,
+			to_warehouse="Stores - TPC",
+			company="Test PCV Company",
+			posting_date="2021-03-15",
+		)
+
+		pcv = self.make_period_closing_voucher(posting_date="2021-03-31", submit=False)
+		sce = self.make_completed_stock_closing_entry(pcv.period_start_date, pcv.period_end_date)
+
+		make_stock_entry(
+			item_code=item.name,
+			qty=5,
+			rate=100,
+			to_warehouse="Stores - TPC",
+			company="Test PCV Company",
+			posting_date="2021-05-01",
+		)
+
+		pcv.reload()
+		self.assertRaisesRegex(frappe.ValidationError, "Regenerate", pcv.submit)
+
+		self.rebuild_stock_closing_balance(sce)
+		pcv.reload()
+		pcv.submit()
+		self.assertEqual(pcv.docstatus, 1)
+
+	def make_completed_stock_closing_entry(self, from_date, to_date):
+		from unittest.mock import patch
+
+		sce = frappe.get_doc(
+			{
+				"doctype": "Stock Closing Entry",
+				"company": "Test PCV Company",
+				"from_date": from_date,
+				"to_date": to_date,
+			}
+		).insert()
+
+		with patch("erpnext.stock.doctype.stock_closing_entry.stock_closing_entry.enqueue"):
+			sce.submit()
+
+		sce.create_stock_closing_balance_entries()
+		sce.db_set("status", "Completed")
+		return sce
+
+	def rebuild_stock_closing_balance(self, sce):
+		sce.remove_stock_closing()
+		sce.create_stock_closing_balance_entries()
+		sce.db_set("status", "Completed")
+
 	def make_period_closing_voucher(self, posting_date, submit=True):
 		surplus_account = create_account()
 		cost_center = create_cost_center("Test Cost Center 1")

erpnext/accounts/doctype/sales_invoice/sales_invoice.py

@@ -1165,6 +1165,7 @@ class SalesInvoice(SellingController):
 		child_tables = {
 			"items": ("income_account", "expense_account", "discount_account"),
 			"taxes": ("account_head",),
+			"payments": ("account",),
 		}
 		self.needs_repost = self.check_if_fields_updated(fields_to_check, child_tables)
 		if self.needs_repost:

erpnext/accounts/doctype/sales_invoice_payment/sales_invoice_payment.json

@@ -1,5 +1,6 @@
 {
  "actions": [],
+ "allow_bulk_edit": 1,
  "creation": "2016-05-08 23:49:38.842621",
  "doctype": "DocType",
  "editable_grid": 1,
@@ -17,6 +18,7 @@
  ],
  "fields": [
   {
+   "allow_on_submit": 1,
    "fieldname": "mode_of_payment",
    "fieldtype": "Link",
    "in_list_view": 1,
@@ -39,6 +41,7 @@
    "fieldtype": "Column Break"
   },
   {
+   "allow_on_submit": 1,
    "fieldname": "account",
    "fieldtype": "Link",
    "label": "Account",
@@ -47,6 +50,7 @@
    "read_only": 1
   },
   {
+   "allow_on_submit": 1,
    "fetch_from": "mode_of_payment.type",
    "fieldname": "type",
    "fieldtype": "Read Only",
@@ -85,7 +89,7 @@
  ],
  "istable": 1,
  "links": [],
- "modified": "2026-02-16 20:46:34.592604",
+ "modified": "2026-07-29 16:44:54.482826",
  "modified_by": "Administrator",
  "module": "Accounts",
  "name": "Sales Invoice Payment",

erpnext/accounts/doctype/subscription/subscription.py

@@ -278,6 +278,9 @@ class Subscription(Document):
 		"""
 		Sets the status of the `Subscription`
 		"""
+		if self.status == STATUS_CANCELLED:
+			return
+
 		self._set_current_invoice_dates()
 		if self.is_trialling():
 			self.status = STATUS_TRIALING
@@ -673,7 +676,7 @@ class Subscription(Document):
 
 		if self.cancel_at_period_end and (
 			getdate(posting_date) >= getdate(self.next_billing_period_end)
-			or getdate(posting_date) >= getdate(self.end_date)
+			or (self.end_date and getdate(posting_date) >= getdate(self.end_date))
 		):
 			self.cancel_subscription()
 

erpnext/accounts/doctype/subscription/test_subscription.py

@@ -779,6 +779,38 @@ class TestSubscription(ERPNextTestSuite):
 		subscription.reload()
 		self.assertEqual(subscription.status, "Active")
 
+	def test_cancelled_subscription_stays_cancelled_after_payment_and_reprocess(self):
+		# https://github.com/frappe/erpnext/issues/57761
+		subscription = create_subscription(
+			start_date=nowdate(),
+			generate_invoice_at="Prepaid (bill at period start)",
+			submit_invoice=1,
+			cancel_at_period_end=1,
+		)
+		subscription.process(posting_date=nowdate())
+		invoice = subscription.get_current_invoice()
+		self.assertGreater(invoice.outstanding_amount, 0)
+
+		subscription.cancel_subscription()
+		self.assertEqual(subscription.status, "Cancelled")
+		cancelation_date = getdate(subscription.cancelation_date)
+		self.assertIsNotNone(cancelation_date)
+
+		payment_entry = get_payment_entry(invoice.doctype, invoice.name, bank_account="_Test Bank - _TC")
+		payment_entry.reference_no = "12345"
+		payment_entry.reference_date = nowdate()
+		payment_entry.submit()
+
+		subscription.reload()
+		self.assertEqual(subscription.status, "Cancelled")
+		self.assertEqual(getdate(subscription.cancelation_date), cancelation_date)
+
+		invoice_count = len(subscription.invoices)
+		subscription.process()
+		subscription.reload()
+		self.assertEqual(subscription.status, "Cancelled")
+		self.assertEqual(len(subscription.invoices), invoice_count)
+
 	def test_first_invoice_generated_on_create_for_prepaid(self):
 		subscription = create_subscription(
 			start_date=nowdate(),

erpnext/accounts/party.py

@@ -865,9 +865,11 @@ def validate_account_party_type(self):
 
 
 def get_dashboard_info(party_type, party, loyalty_program=None):
-	current_fiscal_year = get_fiscal_year(nowdate(), as_dict=True)
-
 	doctype = "Sales Invoice" if party_type == "Customer" else "Purchase Invoice"
+	if not frappe.has_permission(doctype, "read"):
+		return None
+
+	current_fiscal_year = get_fiscal_year(nowdate(), as_dict=True)
 
 	companies = frappe.get_list(
 		doctype, filters={"docstatus": 1, party_type.lower(): party}, distinct=1, fields=["company"]

erpnext/accounts/report/purchase_register/purchase_register.py

@@ -553,8 +553,8 @@ def get_invoice_tax_map(invoice_list, invoice_expense_map, expense_accounts, inc
 			else:
 				invoice_expense_map[d.parent][d.account_head] = flt(d.tax_amount)
 		else:
-			invoice_tax_map.setdefault(d.parent, frappe._dict()).setdefault(d.account_head, [])
-			invoice_tax_map[d.parent][d.account_head] = flt(d.tax_amount)
+			invoice_tax_map.setdefault(d.parent, frappe._dict()).setdefault(d.account_head, 0.0)
+			invoice_tax_map[d.parent][d.account_head] += flt(d.tax_amount)
 
 	return invoice_expense_map, invoice_tax_map
 

erpnext/accounts/report/purchase_register/test_purchase_register.py

@@ -47,6 +47,41 @@ class TestPurchaseRegister(ERPNextTestSuite):
 
 		self.assertEqual(labels, sorted([lower, upper], key=str.casefold))
 
+	def test_add_and_deduct_rows_on_one_account_are_netted(self):
+		"""An account head carrying both an Add and a Deduct row must report their net.
+
+		The tax query groups by (parent, account_head, add_deduct_tax), so such an account comes
+		back as two rows. Only one of them survived into the report.
+		"""
+		from erpnext.accounts.doctype.purchase_invoice.test_purchase_invoice import (
+			make_purchase_invoice as make_pi,
+		)
+
+		company = "_Test Company"
+		tax_account = "_Test Account VAT - _TC"
+
+		pi = make_pi(company=company, do_not_save=True)
+		for add_deduct, amount in (("Add", 10), ("Deduct", 4)):
+			pi.append(
+				"taxes",
+				{
+					"charge_type": "Actual",
+					"account_head": tax_account,
+					"description": "VAT",
+					"category": "Total",
+					"add_deduct_tax": add_deduct,
+					"tax_amount": amount,
+					"cost_center": "Main - _TC",
+				},
+			)
+		pi.save()
+		pi.submit()
+
+		filters = frappe._dict(company=company, from_date=add_months(today(), -1), to_date=today())
+		row = next(r for r in execute(filters)[1] if r.get("voucher_no") == pi.name)
+
+		self.assertEqual(flt(row.get(frappe.scrub(tax_account))), 6.0)
+
 	def test_purchase_register_ignores_tax_rows_from_other_doctype(self):
 		filters = frappe._dict(company="_Test Company 6", from_date=add_months(today(), -1), to_date=today())
 

erpnext/accounts/report/sales_payment_summary/sales_payment_summary.py

@@ -3,7 +3,7 @@
 
 import frappe
 from frappe import _
-from frappe.query_builder.functions import Coalesce, Max, Sum
+from frappe.query_builder.functions import Coalesce, Max, Min, Sum
 from frappe.utils import cstr
 
 
@@ -128,25 +128,47 @@ def get_pos_invoice_data(filters):
 	sip = frappe.qb.DocType("Sales Invoice Payment")
 	si = frappe.qb.DocType("Sales Invoice")
 
-	# t1: one row per invoice with the summed item base_total. warehouse/cost_center are line-level and
-	# not grouped, so they are arbitrary per invoice -- Max() makes that pick deterministic and valid on
-	# Postgres (item_code was selected but never consumed downstream, so it is dropped).
-	t1 = (
+	# t1: one row per invoice with the summed item base_total. warehouse and cost_center describe an
+	# item line, not the invoice, and an invoice may carry several. warehouse then becomes an outer
+	# grouping key below, so which line wins decides how rows are partitioned and what each row totals
+	# -- not merely which label is shown. Max() over text is a sort, and MariaDB (case-folding) and
+	# PostgreSQL (byte order) resolve it differently, so take both off one real line instead.
+	# The representative is the first line the user entered: Min(idx) is an integer, so the pick is
+	# free of collation and is meaningful, rather than turning on an unrelated hash-named row.
+	grouped_items = (
 		frappe.qb.from_(sii)
-		.select(
-			sii.parent,
-			Sum(sii.amount).as_("base_total"),
-			Max(sii.warehouse).as_("warehouse"),
-			Max(sii.cost_center).as_("cost_center"),
-		)
+		.select(sii.parent, Sum(sii.amount).as_("base_total"), Min(sii.idx).as_("representative_idx"))
 		.groupby(sii.parent)
+	).as_("grouped_items")
+	representative_item = frappe.qb.DocType("Sales Invoice Item").as_("representative_item")
+	t1 = (
+		frappe.qb.from_(grouped_items)
+		.inner_join(representative_item)
+		.on(
+			(representative_item.parent == grouped_items.parent)
+			& (representative_item.idx == grouped_items.representative_idx)
+		)
+		.select(
+			grouped_items.parent,
+			grouped_items.base_total,
+			representative_item.warehouse,
+			representative_item.cost_center,
+		)
 	)
 
-	# t3: mode_of_payment per invoice (arbitrary across an invoice's payment lines -> Max() to be valid)
+	# t3: mode_of_payment per invoice, from one real payment line for the same reason
+	grouped_payments = (
+		frappe.qb.from_(sip).select(sip.parent, Min(sip.idx).as_("representative_idx")).groupby(sip.parent)
+	).as_("grouped_payments")
+	representative_payment = frappe.qb.DocType("Sales Invoice Payment").as_("representative_payment")
 	t3 = (
-		frappe.qb.from_(sip)
-		.select(sip.parent, Max(sip.mode_of_payment).as_("mode_of_payment"))
-		.groupby(sip.parent)
+		frappe.qb.from_(grouped_payments)
+		.inner_join(representative_payment)
+		.on(
+			(representative_payment.parent == grouped_payments.parent)
+			& (representative_payment.idx == grouped_payments.representative_idx)
+		)
+		.select(grouped_payments.parent, representative_payment.mode_of_payment.as_("mode_of_payment"))
 	)
 
 	# a: invoice-level aggregates. Grouped by the primary key (si.name), so the other plain si columns

erpnext/accounts/report/sales_payment_summary/test_sales_payment_summary.py

@@ -54,6 +54,53 @@ class TestSalesPaymentSummary(ERPNextTestSuite):
 		self.assertIn("Credit Card", next(iter(mop.values())))
 		self.assertNotIn("Cash", next(iter(mop.values())))
 
+	def test_pos_invoice_warehouse_and_cost_center_come_from_one_item(self):
+		"""The reported warehouse and cost centre must belong to the same item line.
+
+		They describe a line, not the invoice, and an invoice can carry several. Aggregating each
+		on its own can report a warehouse from one line beside a cost centre from another -- a pair
+		that was never posted. The warehouse is also an outer grouping key, so the pick decides how
+		rows are partitioned and what each one totals, not just what is displayed.
+		"""
+		from erpnext.stock.doctype.item.test_item import make_item
+		from erpnext.stock.doctype.warehouse.test_warehouse import create_warehouse
+
+		low_warehouse = create_warehouse("_Test POS Summary AAA")
+		high_warehouse = create_warehouse("_Test POS Summary ZZZ")
+		second_item = make_item("_Test POS Summary Second Item", {"is_stock_item": 0}).name
+
+		si = create_sales_invoice_record()
+		si.is_pos = 1
+		# cross the two picks: the higher warehouse is on the line with the lower cost centre, so an
+		# independently aggregated pair cannot belong to either line
+		si.items[0].warehouse = high_warehouse
+		si.items[0].cost_center = "Main - _TC"
+		si.append(
+			"items",
+			{
+				"item_code": second_item,
+				"qty": 1,
+				"rate": 5000,
+				"income_account": "Sales - _TC",
+				"expense_account": "Cost of Goods Sold - _TC",
+				"warehouse": low_warehouse,
+				"cost_center": "Sub - _TC",
+			},
+		)
+		si.append("payments", {"mode_of_payment": "Cash", "account": "_Test Cash - _TC", "amount": 15000})
+		si.insert()
+		si.submit()
+
+		posted = {(row.warehouse, row.cost_center) for row in si.items}
+		self.assertGreater(len(posted), 1, "fixture must post more than one distinct pair")
+
+		rows = get_pos_invoice_data(get_filters())
+		reported = [r for r in rows if r.get("warehouse") in {w for w, _ in posted}]
+		self.assertTrue(reported)
+
+		for row in reported:
+			self.assertIn((row["warehouse"], row["cost_center"]), posted)
+
 	def test_get_mode_of_payments_details(self):
 		filters = get_filters()
 

erpnext/accounts/services/exchange_gain_loss.py

@@ -195,7 +195,7 @@ def make_exchange_gain_loss_journal(
 
 def is_payable_account(reference_doctype: str, account: str) -> bool:
 	if reference_doctype == "Purchase Invoice" or (
-		reference_doctype == "Journal Entry"
+		reference_doctype in ("Journal Entry", "Payment Entry")
 		and frappe.get_cached_value("Account", account, "account_type") == "Payable"
 	):
 		return True

erpnext/accounts/utils.py

@@ -2379,13 +2379,16 @@ class QueryPaymentLedger:
 				)
 
 		# build query for voucher amount
-		query_voucher_amount = (
+		# account is grouped, not aggregated: it is a join key against the outstanding CTE below, and
+		# it fixes the currency the amounts are summed in. The two CTEs aggregate over different row
+		# sets, so two Max() picks could disagree and the join would silently miss, leaving the
+		# outstanding NULL. posting_date/due_date are dates, so Max() there cannot depend on
+		# collation. cost_center and remarks are free text that genuinely varies per row, so they
+		# come off one real row instead -- see representative below.
+		grouped_voucher_amount = (
 			qb.from_(ple)
 			.select(
-				# columns that are constant per (voucher_type, voucher_no, party_type, party) are
-				# wrapped in Max() so the query is valid on postgres (which, unlike MariaDB, requires
-				# every non-aggregated column to be grouped or aggregated)
-				Max(ple.account).as_("account"),
+				ple.account,
 				ple.voucher_type,
 				ple.voucher_no,
 				ple.party_type,
@@ -2393,25 +2396,47 @@ class QueryPaymentLedger:
 				Max(ple.posting_date).as_("posting_date"),
 				Max(ple.due_date).as_("due_date"),
 				Max(ple.account_currency).as_("currency"),
-				Max(ple.cost_center).as_("cost_center"),
 				Sum(ple.amount).as_("amount"),
 				Sum(ple.amount_in_account_currency).as_("amount_in_account_currency"),
-				Max(ple.remarks).as_("remarks"),
+				Min(ple.name).as_("representative"),
 			)
 			.where(ple.delinked == 0)
 			.where(Criterion.all(filter_on_voucher_no))
 			.where(Criterion.all(self.common_filter))
 			.where(Criterion.all(self.dimensions_filter))
 			.where(Criterion.all(self.voucher_posting_date))
-			.groupby(ple.voucher_type, ple.voucher_no, ple.party_type, ple.party)
+			.groupby(ple.account, ple.voucher_type, ple.voucher_no, ple.party_type, ple.party)
+		).as_("grouped")
+
+		# Payment Ledger Entry has no autoname rule, so frappe names it by hash -- lower-case, which
+		# keeps Min(name) free of the collation divergence that picking Max() over free text has.
+		representative_ple = qb.DocType("Payment Ledger Entry").as_("representative_ple")
+		query_voucher_amount = (
+			qb.from_(grouped_voucher_amount)
+			.inner_join(representative_ple)
+			.on(representative_ple.name == grouped_voucher_amount.representative)
+			.select(
+				grouped_voucher_amount.account,
+				grouped_voucher_amount.voucher_type,
+				grouped_voucher_amount.voucher_no,
+				grouped_voucher_amount.party_type,
+				grouped_voucher_amount.party,
+				grouped_voucher_amount.posting_date,
+				grouped_voucher_amount.due_date,
+				grouped_voucher_amount.currency,
+				grouped_voucher_amount.amount,
+				grouped_voucher_amount.amount_in_account_currency,
+				representative_ple.cost_center.as_("cost_center"),
+				representative_ple.remarks.as_("remarks"),
+			)
 		)
 
 		# build query for voucher outstanding
 		query_voucher_outstanding = (
 			qb.from_(ple)
 			.select(
-				# Max() on columns constant per group keeps this valid on postgres (see above)
-				Max(ple.account).as_("account"),
+				# grouped, not aggregated: this is the other side of the join key -- see above
+				ple.account,
 				ple.against_voucher_type.as_("voucher_type"),
 				ple.against_voucher_no.as_("voucher_no"),
 				ple.party_type,
@@ -2425,7 +2450,7 @@ class QueryPaymentLedger:
 			.where(ple.delinked == 0)
 			.where(Criterion.all(filter_on_against_voucher_no))
 			.where(Criterion.all(self.common_filter))
-			.groupby(ple.against_voucher_type, ple.against_voucher_no, ple.party_type, ple.party)
+			.groupby(ple.account, ple.against_voucher_type, ple.against_voucher_no, ple.party_type, ple.party)
 		)
 
 		# build CTE for combining voucher amount and outstanding

erpnext/buying/doctype/purchase_order/test_purchase_order.py

@@ -216,6 +216,21 @@ class TestPurchaseOrder(ERPNextTestSuite):
 		po2.items[0].qty = 110
 		self.assertRaises(OverAllowanceError, po2.submit)
 
+		# Stock over-delivery role must not bypass over-ordering against Material Request.
+		with self.change_settings(
+			"Stock Settings", {"role_allowed_to_over_deliver_receive": "Stock Manager"}
+		):
+			test_user = frappe.get_doc("User", "test@example.com")
+			test_user.add_roles("Stock Manager")
+
+			mr3 = make_material_request(qty=100)
+			po3 = make_purchase_order(mr3.name)
+			po3.supplier = "_Test Supplier"
+			po3.items[0].qty = 110
+			with self.set_user("test@example.com"):
+				po3.flags.ignore_permissions = True
+				self.assertRaises(OverAllowanceError, po3.submit)
+
 		# cleanup
 		frappe.db.set_single_value("Buying Settings", "over_order_allowance", 0)
 		frappe.db.set_single_value("Stock Settings", "over_delivery_receipt_allowance", 0)
@@ -1044,6 +1059,8 @@ class TestPurchaseOrder(ERPNextTestSuite):
 		# self.assertEqual(po.payment_terms_template, pi.payment_terms_template)
 		compare_payment_schedules(self, po, pi)
 
+	@ERPNextTestSuite.change_settings("Selling Settings", {"maintain_same_sales_rate": 1})
+	@ERPNextTestSuite.change_settings("Buying Settings", {"maintain_same_rate": 1})
 	def test_internal_transfer_flow(self):
 		from erpnext.accounts.doctype.cost_center.test_cost_center import create_cost_center
 		from erpnext.accounts.doctype.sales_invoice.mapper import (
@@ -1055,9 +1072,6 @@ class TestPurchaseOrder(ERPNextTestSuite):
 		)
 		from erpnext.stock.doctype.delivery_note.mapper import make_inter_company_purchase_receipt
 
-		frappe.db.set_single_value("Selling Settings", "maintain_same_sales_rate", 1)
-		frappe.db.set_single_value("Buying Settings", "maintain_same_rate", 1)
-
 		prepare_data_for_internal_transfer()
 		supplier = "_Test Internal Supplier 2"
 
@@ -1496,6 +1510,7 @@ class TestPurchaseOrder(ERPNextTestSuite):
 		self.assertEqual(pi_2.status, "Paid")
 		self.assertEqual(po.status, "Completed")
 
+	@ERPNextTestSuite.change_settings("Buying Settings", {"maintain_same_rate": 0})
 	def test_purchase_order_over_billing_missing_item(self):
 		item1 = make_item(
 			"_Test Item for Overbilling",

erpnext/buying/report/requested_items_to_order_and_receive/requested_items_to_order_and_receive.py

@@ -51,7 +51,6 @@ def get_data(filters):
 			mr_item.item_code.as_("item_code"),
 			Sum(Coalesce(mr_item.qty, 0)).as_("qty"),
 			Sum(Coalesce(mr_item.stock_qty, 0)).as_("stock_qty"),
-			Max(Coalesce(mr_item.uom, "")).as_("uom"),
 			Max(Coalesce(mr_item.stock_uom, "")).as_("stock_uom"),
 			Sum(Coalesce(mr_item.ordered_qty, 0)).as_("ordered_qty"),
 			Sum(Coalesce(mr_item.received_qty, 0)).as_("received_qty"),
@@ -60,8 +59,6 @@ def get_data(filters):
 			),
 			Sum(Coalesce(mr_item.received_qty, 0)).as_("received_qty"),
 			(Sum(Coalesce(mr_item.stock_qty, 0)) - Sum(Coalesce(mr_item.ordered_qty, 0))).as_("qty_to_order"),
-			Max(mr_item.item_name).as_("item_name"),
-			Max(mr_item.description).as_("description"),
 			Max(mr.company).as_("company"),
 		)
 		.where(
@@ -75,8 +72,34 @@ def get_data(filters):
 	query = get_conditions(filters, query, mr, mr_item)  # add conditional conditions
 
 	query = query.groupby(mr.name, mr_item.item_code).orderby(Max(mr.transaction_date), Max(mr.schedule_date))
-	data = query.run(as_dict=True)
-	return data
+	rows = query.run(as_dict=True)
+	apply_representative_lines(rows)
+	return rows
+
+
+def apply_representative_lines(rows):
+	"""Fill item_name/description/uom from one real Material Request Item line per group.
+
+	All three are editable per line, so a request listing the same item twice holds several values
+	per group. Aggregating them sorts text, and MariaDB folds case while PostgreSQL orders by byte
+	value, so the engines pick differently. Take the first line by idx.
+	"""
+	material_requests = list({row.material_request for row in rows})
+	representative = {}
+	if material_requests:
+		for line in frappe.get_all(
+			"Material Request Item",
+			filters={"parent": ("in", material_requests), "docstatus": 1},
+			fields=["parent", "item_code", "item_name", "description", "uom"],
+			order_by="idx",
+		):
+			representative.setdefault((line.parent, line.item_code), line)
+
+	for row in rows:
+		line = representative.get((row.material_request, row.item_code))
+		row.item_name = line.item_name if line else None
+		row.description = line.description if line else None
+		row.uom = line.uom if line else ""
 
 
 def get_conditions(filters, query, mr, mr_item):

erpnext/controllers/queries.py

@@ -73,14 +73,17 @@ def employee_query(
 		.where(Criterion.any(search_conditions))
 		.orderby(
 			Case()
-			.when(Locate(txt_no_percent, Employee.name) > 0, Locate(txt_no_percent, Employee.name))
+			.when(
+				Locate(Lower(txt_no_percent), Lower(Employee.name)) > 0,
+				Locate(Lower(txt_no_percent), Lower(Employee.name)),
+			)
 			.else_(99999)
 		)
 		.orderby(
 			Case()
 			.when(
-				Locate(txt_no_percent, Employee.employee_name) > 0,
-				Locate(txt_no_percent, Employee.employee_name),
+				Locate(Lower(txt_no_percent), Lower(Employee.employee_name)) > 0,
+				Locate(Lower(txt_no_percent), Lower(Employee.employee_name)),
 			)
 			.else_(99999)
 		)
@@ -136,17 +139,28 @@ def lead_query(
 		query.where(Lead.docstatus < 2)
 		.where(Lead.status.isnull() | (Lead.status != "Converted"))
 		.where(Criterion.any(search_conditions))
-		.orderby(
-			Case().when(Locate(txt_no_percent, Lead.name) > 0, Locate(txt_no_percent, Lead.name)).else_(99999)
-		)
 		.orderby(
 			Case()
-			.when(Locate(txt_no_percent, Lead.lead_name) > 0, Locate(txt_no_percent, Lead.lead_name))
+			.when(
+				Locate(Lower(txt_no_percent), Lower(Lead.name)) > 0,
+				Locate(Lower(txt_no_percent), Lower(Lead.name)),
+			)
 			.else_(99999)
 		)
 		.orderby(
 			Case()
-			.when(Locate(txt_no_percent, Lead.company_name) > 0, Locate(txt_no_percent, Lead.company_name))
+			.when(
+				Locate(Lower(txt_no_percent), Lower(Lead.lead_name)) > 0,
+				Locate(Lower(txt_no_percent), Lower(Lead.lead_name)),
+			)
+			.else_(99999)
+		)
+		.orderby(
+			Case()
+			.when(
+				Locate(Lower(txt_no_percent), Lower(Lead.company_name)) > 0,
+				Locate(Lower(txt_no_percent), Lower(Lead.company_name)),
+			)
 			.else_(99999)
 		)
 		.orderby(Lead.idx, order=Order.desc)
@@ -387,7 +401,12 @@ def bom(
 		.where(BOM.is_active == 1)
 		.where(BOM[searchfield].like(f"%{txt}%"))
 		.orderby(
-			Case().when(Locate(txt_no_percent, BOM.name) > 0, Locate(txt_no_percent, BOM.name)).else_(99999)
+			Case()
+			.when(
+				Locate(Lower(txt_no_percent), Lower(BOM.name)) > 0,
+				Locate(Lower(txt_no_percent), Lower(BOM.name)),
+			)
+			.else_(99999)
 		)
 		.orderby(BOM.idx, order=Order.desc)
 		.orderby(BOM.name)

erpnext/controllers/sales_and_purchase_return.py

@@ -160,10 +160,28 @@ def validate_returned_items(doc):
 				):
 					frappe.throw(_("Warehouse is mandatory"))
 
-			items_returned = True
+			if doc.doctype in (
+				"Purchase Invoice",
+				"Purchase Receipt",
+				"Subcontracting Receipt",
+				"Sales Invoice",
+				"Delivery Note",
+				"POS Invoice",
+			):
+				if flt(d.qty) < 0 or flt(d.get("received_qty")) < 0:
+					items_returned = True
+			else:
+				items_returned = True
 
 		elif d.item_name:
-			items_returned = True
+			if doc.doctype in ("Purchase Invoice", "Purchase Receipt", "Subcontracting Receipt"):
+				# No item_code here means no linked Item, so there's no accepted/rejected
+				# split to speak of - received_qty isn't a meaningful independent signal.
+				# Only a negative qty (i.e. a real negative billing amount) counts.
+				if flt(d.qty) < 0:
+					items_returned = True
+			else:
+				items_returned = True
 
 	if not items_returned:
 		frappe.throw(_("At least one item should be entered with negative quantity in return document"))

erpnext/controllers/status_updater.py

@@ -446,11 +446,12 @@ class StatusUpdater(Document):
 			else (0, {}, None, None)
 		)
 
-		role_allowed_to_over_deliver_receive = frappe.get_single_value(
-			"Stock Settings", "role_allowed_to_over_deliver_receive"
-		)
-		role_allowed_to_over_bill = frappe.get_single_value("Accounts Settings", "role_allowed_to_over_bill")
-		role = role_allowed_to_over_deliver_receive if qty_or_amount == "qty" else role_allowed_to_over_bill
+		role = None
+		if qty_or_amount == "qty":
+			if args.get("overflow_type") in ("delivery", "receipt"):
+				role = frappe.get_single_value("Stock Settings", "role_allowed_to_over_deliver_receive")
+		else:
+			role = frappe.get_single_value("Accounts Settings", "role_allowed_to_over_bill")
 
 		overflow_percent = (
 			(item[args["target_field"]] - item[args["target_ref_field"]]) / item[args["target_ref_field"]]

erpnext/controllers/tests/test_queries.py

@@ -29,6 +29,27 @@ class TestQueries(ERPNextTestSuite):
 		self.assertGreaterEqual(len(query(txt="_Test Lead")), 4)
 		self.assertEqual(len(query(txt="_Test Lead 4")), 1)
 
+	def test_lead_query_ranking_is_case_insensitive(self):
+		"""A match at the start must rank first whatever its case.
+
+		The filter uses .like(), which frappe renders as ILIKE on PostgreSQL, so both leads match.
+		Ranking used a bare Locate(), which becomes case-sensitive strpos() there: the upper-cased
+		lead scores no match, falls back to 99999 and sorts last, while MariaDB's case-insensitive
+		LOCATE ranks it first. Same query, different order -- and a different page when page_len is
+		small enough to cut between them.
+		"""
+		early, late = "ZZABCD Ranking Lead", "Ranking Lead zzabcd"
+		for lead_name in (early, late):
+			if not frappe.db.exists("Lead", {"lead_name": lead_name}):
+				frappe.get_doc({"doctype": "Lead", "lead_name": lead_name}).insert()
+
+		query = add_default_params(queries.lead_query, "Lead")
+		names = [row[1] for row in query(txt="zzabcd")]
+
+		self.assertIn(early, names)
+		self.assertIn(late, names)
+		self.assertLess(names.index(early), names.index(late))
+
 	def test_item_query(self):
 		query = add_default_params(queries.item_query, "Item")
 

erpnext/controllers/tests/test_sales_and_purchase_return.py

@@ -37,3 +37,76 @@ class TestSalesAndPurchaseReturn(ERPNextTestSuite):
 
 		self.assertEqual(return_dn.is_return, 1)
 		self.assertEqual(return_dn.items[0].qty, -5)
+
+	def test_purchase_invoice_zero_qty_return_is_rejected(self):
+		# A return with every item at qty 0 moves no stock and no value, so it must be
+		# rejected the same way a return with no items at all would be.
+		from erpnext.accounts.doctype.purchase_invoice.test_purchase_invoice import make_purchase_invoice
+
+		pi = make_purchase_invoice(qty=10)
+		self.addCleanup(self._cancel_and_delete, "Purchase Invoice", pi.name)
+
+		return_pi = make_purchase_invoice(
+			is_return=1,
+			return_against=pi.name,
+			qty=0,
+			do_not_save=True,
+		)
+
+		self.assertRaises(frappe.ValidationError, return_pi.save)
+
+	def test_purchase_invoice_item_name_only_zero_qty_return_is_rejected(self):
+		# Item Code is not mandatory on Purchase Invoice Item - a row can have only an
+		# item_name (e.g. a free-text/non-stock line). Such rows fall through to the
+		# item_name-only branch, which must also reject an all-zero-qty return instead
+		# of unconditionally treating the row as returned.
+		from erpnext.accounts.doctype.purchase_invoice.test_purchase_invoice import make_purchase_invoice
+
+		pi = make_purchase_invoice(item_name="_Test Item", qty=10, do_not_submit=True)
+		pi.items[0].item_code = ""
+		pi.save()
+		pi.submit()
+		self.addCleanup(self._cancel_and_delete, "Purchase Invoice", pi.name)
+
+		return_pi = make_purchase_invoice(
+			item_name="_Test Item",
+			is_return=1,
+			return_against=pi.name,
+			qty=0,
+			do_not_save=True,
+		)
+		return_pi.items[0].item_code = ""
+
+		self.assertRaises(frappe.ValidationError, return_pi.save)
+
+	def test_delivery_note_zero_qty_return_is_rejected(self):
+		# A return with every item at qty 0 moves no stock and no value, so it must be
+		# rejected the same way a return with no items at all would be.
+		from erpnext.stock.doctype.delivery_note.mapper import make_sales_return
+		from erpnext.stock.doctype.delivery_note.test_delivery_note import create_delivery_note
+		from erpnext.stock.doctype.stock_entry.stock_entry_utils import make_stock_entry
+
+		se = make_stock_entry(item_code="_Test Item", target="_Test Warehouse - _TC", qty=20, basic_rate=100)
+		self.addCleanup(self._cancel_and_delete, "Stock Entry", se.name)
+
+		dn = create_delivery_note(qty=5)
+		self.addCleanup(self._cancel_and_delete, "Delivery Note", dn.name)
+
+		return_dn = make_sales_return(dn.name)
+		return_dn.items[0].qty = 0
+
+		self.assertRaises(frappe.ValidationError, return_dn.insert)
+
+	def test_sales_invoice_zero_qty_return_is_rejected(self):
+		# Same rule for a standalone (non stock-affecting) Sales Invoice return: qty 0 on
+		# every row must be rejected, not silently accepted as a no-op credit note.
+		from erpnext.accounts.doctype.sales_invoice.test_sales_invoice import create_sales_invoice
+		from erpnext.controllers.sales_and_purchase_return import make_return_doc
+
+		si = create_sales_invoice(qty=10)
+		self.addCleanup(self._cancel_and_delete, "Sales Invoice", si.name)
+
+		return_si = make_return_doc(si.doctype, si.name)
+		return_si.items[0].qty = 0
+
+		self.assertRaises(frappe.ValidationError, return_si.save)

erpnext/controllers/trends.py

@@ -376,6 +376,41 @@ def get_period_month_ranges(period, fiscal_year):
 	return period_month_ranges
 
 
+def quotation_party_name_expr():
+	"""Resolve a Quotation's party label from its dynamic link, mirroring set_customer_name()."""
+	customer_branch = (
+		"when t1.quotation_to = 'Customer' then "
+		"(select c.customer_name from `tabCustomer` c where c.name = t1.party_name)"
+	)
+	lead_branch = (
+		"when t1.quotation_to = 'Lead' then "
+		"(select coalesce(nullif(l.company_name, ''), l.lead_name) from `tabLead` l "
+		"where l.name = t1.party_name)"
+	)
+	prospect_branch = "when t1.quotation_to = 'Prospect' then t1.party_name"
+	branches = [customer_branch, lead_branch, prospect_branch]
+	# CRM Deal ships with the CRM app; skip the branch when its table is absent
+	if frappe.db.table_exists("CRM Deal"):
+		branches.append(
+			"when t1.quotation_to = 'CRM Deal' then "
+			"(select d.organization from `tabCRM Deal` d where d.name = t1.party_name)"
+		)
+
+	return "case " + " ".join(branches) + " end"
+
+
+def quotation_territory_expr():
+	"""Only Customer and Lead carry a territory; other party types have none."""
+	return (
+		"case "
+		"when t1.quotation_to = 'Customer' then "
+		"(select c.territory from `tabCustomer` c where c.name = t1.party_name) "
+		"when t1.quotation_to = 'Lead' then "
+		"(select l.territory from `tabLead` l where l.name = t1.party_name) "
+		"end"
+	)
+
+
 def based_wise_columns_query(based_on, trans):
 	based_on_details = {}
 
@@ -385,12 +420,14 @@ def based_wise_columns_query(based_on, trans):
 			{"label": _("Item"), "fieldtype": "Link", "options": "Item", "width": 120, "fieldname": "item"},
 			{"label": _("Item Name"), "fieldtype": "Data", "width": 120, "fieldname": "item_name"},
 		]
-		# item_name is an editable per-line field, not functionally dependent on item_code, so it
-		# is aggregated (one row per item_code) rather than added to GROUP BY (which would split
-		# the row and change the MariaDB row count). See get_data's group-by query.
-		based_on_details["based_on_select"] = "t2.item_code, Max(t2.item_name) as item_name,"
-		based_on_details["based_on_group_by"] = "t2.item_code"
-		based_on_details["addl_tables"] = ""
+		# item_name is stored per line and editable, so it is not functionally dependent on item_code
+		# and Max() over it is a sort -- which MariaDB and PostgreSQL resolve differently. Read it
+		# from the Item master instead: that IS functionally dependent on the grouped item_code, so
+		# it can be grouped without splitting rows and is identical on both engines by construction.
+		based_on_details["based_on_select"] = "t2.item_code, item_master.item_name as item_name,"
+		based_on_details["based_on_group_by"] = "t2.item_code, item_master.item_name"
+		based_on_details["addl_tables"] = ",`tabItem` item_master"
+		based_on_details["addl_tables_relational_cond"] = " and t2.item_code = item_master.name"
 
 	elif based_on == "Item Group":
 		based_on_details["based_on_cols"] = [
@@ -425,9 +462,17 @@ def based_wise_columns_query(based_on, trans):
 					"fieldname": "territory",
 				},
 			]
-			based_on_details[
-				"based_on_select"
-			] = "t1.party_name, Max(t1.customer_name) as customer_name, Max(t1.territory) as territory,"
+			# a Quotation's party_name is a dynamic link, so no single master can be joined. Resolve
+			# it through the quotation_to discriminator, mirroring Quotation.set_customer_name, and
+			# group by it too: two parties of different types can share a name, and merging them
+			# under one row was never right. Correlated only on grouped columns, so the query stays
+			# valid under GROUP BY and free of any text sort.
+			based_on_details["based_on_select"] = (
+				f"t1.party_name, {quotation_party_name_expr()} as customer_name, "
+				f"{quotation_territory_expr()} as territory,"
+			)
+			based_on_details["based_on_group_by"] = "t1.party_name, t1.quotation_to"
+			based_on_details["addl_tables"] = ""
 		else:
 			based_on_details["based_on_cols"] = [
 				{
@@ -451,13 +496,19 @@ def based_wise_columns_query(based_on, trans):
 					"fieldname": "territory",
 				},
 			]
+			# customer_name and territory are stored per transaction and editable, so they are not
+			# functionally dependent on the customer and Max() over them is a text sort, which the
+			# engines resolve differently. The Customer master's values ARE dependent on the grouped
+			# key, so they can be grouped without splitting rows and agree on both engines.
+			based_on_details["based_on_select"] = (
+				"t1.customer, customer_master.customer_name as customer_name, "
+				"customer_master.territory as territory,"
+			)
 			based_on_details[
-				"based_on_select"
-			] = "t1.customer, Max(t1.customer_name) as customer_name, Max(t1.territory) as territory,"
-		# territory (and customer_name) are not functionally dependent on the customer key, so they
-		# are aggregated rather than grouped — one row per customer, matching the prior MariaDB output.
-		based_on_details["based_on_group_by"] = "t1.party_name" if trans == "Quotation" else "t1.customer"
-		based_on_details["addl_tables"] = ""
+				"based_on_group_by"
+			] = "t1.customer, customer_master.customer_name, customer_master.territory"
+			based_on_details["addl_tables"] = ",`tabCustomer` customer_master"
+			based_on_details["addl_tables_relational_cond"] = " and t1.customer = customer_master.name"
 
 	elif based_on == "Customer Group":
 		based_on_details["based_on_cols"] = [
@@ -490,14 +541,12 @@ def based_wise_columns_query(based_on, trans):
 				"fieldname": "supplier_group",
 			},
 		]
-		# supplier_name is a stored per-transaction field (not functionally dependent on supplier), so
-		# it is aggregated to keep one row per supplier — matching the prior MariaDB output, which grouped
-		# by t1.supplier only. supplier_group comes from the joined master and is FD on supplier, so it
-		# stays in GROUP BY (postgres-valid, no row split).
-		based_on_details[
-			"based_on_select"
-		] = "t1.supplier, Max(t1.supplier_name) as supplier_name, t3.supplier_group,"
-		based_on_details["based_on_group_by"] = "t1.supplier, t3.supplier_group"
+		# supplier_name is stored per transaction and editable, so Max() over it is a text sort that
+		# the engines resolve differently. The Supplier master is already joined here as t3 and its
+		# columns are functionally dependent on the grouped supplier, so both can simply be grouped:
+		# no row split, and identical on both engines by construction.
+		based_on_details["based_on_select"] = "t1.supplier, t3.supplier_name, t3.supplier_group,"
+		based_on_details["based_on_group_by"] = "t1.supplier, t3.supplier_name, t3.supplier_group"
 		based_on_details["addl_tables"] = ",`tabSupplier` t3"
 		based_on_details["addl_tables_relational_cond"] = " and t1.supplier = t3.name"
 

erpnext/crm/doctype/opportunity/opportunity.py

@@ -133,6 +133,7 @@ class Opportunity(TransactionBase, CRMNote):
 		self.validate_uom_is_integer("uom", "qty")
 		self.validate_cust_name()
 		self.map_fields()
+		self.validate_qty()
 		self.set_exchange_rate()
 
 		if not self.title:
@@ -143,6 +144,15 @@ class Opportunity(TransactionBase, CRMNote):
 	def on_update(self):
 		self.update_prospect()
 
+	def validate_qty(self):
+		for item in self.items:
+			if flt(item.qty) <= 0:
+				frappe.throw(
+					_("Row #{0}: Quantity must be greater than 0 for Item {1}").format(
+						item.idx, item.item_code
+					)
+				)
+
 	def map_fields(self):
 		for field in self.meta.get_valid_columns():
 			if not self.get(field) and frappe.db.field_exists(self.opportunity_from, field):

erpnext/manufacturing/doctype/blanket_order/blanket_order.py

@@ -120,8 +120,8 @@ class BlanketOrder(Document):
 
 	def validate_item_qty(self):
 		for d in self.items:
-			if flt(d.qty) < 0:
-				frappe.throw(_("Row {0}: Quantity cannot be negative.").format(d.idx))
+			if flt(d.qty) <= 0:
+				frappe.throw(_("Row {0}: Quantity must be greater than zero.").format(d.idx))
 
 
 @frappe.whitelist()
@@ -149,7 +149,11 @@ def make_order(source_name: str):
 		"Blanket Order",
 		source_name,
 		{
-			"Blanket Order": {"doctype": doctype, "postprocess": update_doc},
+			"Blanket Order": {
+				"doctype": doctype,
+				"field_no_map": ["naming_series"],
+				"postprocess": update_doc,
+			},
 			"Blanket Order Item": {
 				"doctype": doctype + " Item",
 				"field_map": {"rate": "blanket_order_rate", "parent": "blanket_order"},

erpnext/manufacturing/doctype/blanket_order/test_blanket_order.py

@@ -25,6 +25,7 @@ class TestBlanketOrder(ERPNextTestSuite):
 		so.submit()
 
 		self.assertEqual(so.doctype, "Sales Order")
+		self.assertNotEqual(so.naming_series, bo.naming_series)
 		self.assertEqual(len(so.get("items")), len(bo.get("items")))
 
 		# check the rate, quantity and updation for the ordered quantity
@@ -50,6 +51,7 @@ class TestBlanketOrder(ERPNextTestSuite):
 		po.submit()
 
 		self.assertEqual(po.doctype, "Purchase Order")
+		self.assertNotEqual(po.naming_series, bo.naming_series)
 		self.assertEqual(len(po.get("items")), len(bo.get("items")))
 
 		# check the rate, quantity and updation for the ordered quantity
@@ -91,6 +93,32 @@ class TestBlanketOrder(ERPNextTestSuite):
 		frappe.db.set_single_value("Buying Settings", "blanket_order_allowance", 10)
 		po.submit()
 
+	@ERPNextTestSuite.change_settings("Selling Settings", {"blanket_order_allowance": 0})
+	@ERPNextTestSuite.change_settings("Buying Settings", {"blanket_order_allowance": 0})
+	@ERPNextTestSuite.change_settings(
+		"Stock Settings",
+		{"over_delivery_receipt_allowance": 10, "role_allowed_to_over_deliver_receive": "Stock Manager"},
+	)
+	def test_stock_over_delivery_role_does_not_bypass_blanket_order_allowance(self):
+		test_user = frappe.get_doc("User", "test@example.com")
+		test_user.add_roles("Stock Manager")
+
+		frappe.clear_cache()
+		for blanket_order_type, doctype, date_field in (
+			("Selling", "Sales Order", "delivery_date"),
+			("Purchasing", "Purchase Order", "schedule_date"),
+		):
+			bo = make_blanket_order(blanket_order_type=blanket_order_type, quantity=100)
+			frappe.flags.args.doctype = doctype
+			order = make_order(bo.name)
+			order.currency = get_company_currency(order.company)
+			setattr(order, date_field, today())
+			order.items[0].qty = 110
+
+			with self.set_user("test@example.com"):
+				order.flags.ignore_permissions = True
+				self.assertRaises(frappe.ValidationError, order.submit)
+
 	def test_blanket_order_over_order_aggregated_across_rows(self):
 		# the over-order check should sum the same item across multiple order rows
 		frappe.db.set_single_value("Selling Settings", "blanket_order_allowance", 0)
@@ -136,6 +164,26 @@ class TestBlanketOrder(ERPNextTestSuite):
 		bo = make_blanket_order(blanket_order_type="Purchasing", supplier=supplier, item_code=item_code)
 		self.assertEqual(bo.items[0].party_item_code, "SUPP-PART-1")
 
+	def test_blanket_order_zero_quantity(self):
+		bo = frappe.new_doc("Blanket Order")
+		bo.blanket_order_type = "Selling"
+		bo.company = "_Test Company"
+		bo.customer = "_Test Customer"
+		bo.from_date = today()
+		bo.to_date = add_months(today(), 12)
+
+		bo.append(
+			"items",
+			{
+				"item_code": "_Test Item",
+				"qty": 0,
+				"rate": 100,
+			},
+		)
+
+		with self.assertRaises(frappe.ValidationError):
+			bo.insert()
+
 
 def make_blanket_order(**args):
 	args = frappe._dict(args)

erpnext/manufacturing/doctype/bom/bom.py

@@ -11,6 +11,7 @@ from frappe.model.document import Document
 from frappe.query_builder import Field
 from frappe.query_builder.functions import Count, IfNull, Max, Min, NullIf, Sum
 from frappe.utils import cint, cstr, flt, get_link_to_form, parse_json
+from frappe.utils.caching import request_cache
 from frappe.website.website_generator import WebsiteGenerator
 
 import erpnext
@@ -1208,7 +1209,65 @@ def _query_bom_items(bom, company, opts):
 	query, group_by = _add_bom_item_columns(query, t, bom, opts, track_semi_finished_goods)
 	# qualify + aggregate idx: bare "idx" is ambiguous across the joined tables and isn't grouped
 	# (idx is unique per BOM item, so Min() preserves the original ordering) — needed for postgres
-	return query.groupby(*group_by).orderby(Min(t.bom_item.idx)).run(as_dict=True)
+	rows = query.groupby(*group_by).orderby(Min(t.bom_item.idx)).run(as_dict=True)
+
+	if not opts.fetch_secondary_items:
+		doctype = "BOM Explosion Item" if cint(opts.fetch_exploded) else "BOM Item"
+		# key only on group-by columns that belong to the line table. stock_uom is grouped from Item
+		# and can differ from the line's stored copy once an item's stock UOM is changed after the
+		# BOM was submitted; keying on it would miss and blank the row. It is functionally dependent
+		# on item_code anyway, so dropping it from the key loses nothing.
+		keys = [field.name for field in group_by if field.table is t.bom_item]
+		_apply_representative_lines(rows, doctype, bom, keys)
+
+	return rows
+
+
+def _line_columns_for(doctype):
+	columns = ["description", "source_warehouse"]
+	if doctype == "BOM Item":
+		# uom only means something beside its own conversion_factor, so they travel together
+		columns += ["uom", "conversion_factor"]
+	return columns
+
+
+def _apply_representative_lines(rows, doctype, bom, keys):
+	"""Fill the line-level columns from a single real BOM line per group.
+
+	They describe a line, not an item, so a BOM listing the same item more than once holds several
+	values per group. Aggregating each independently can pair one line's description with another's
+	warehouse -- or a uom with the wrong conversion_factor -- and Max() over text is a sort, which
+	MariaDB (case-folding) and PostgreSQL (byte order) resolve differently. Take the first by idx.
+	"""
+	repeated = [row for row in rows if (row.pop("line_count", 1) or 1) > 1]
+	if not repeated:
+		return
+
+	columns = _line_columns_for(doctype)
+	representative = _representative_lines(doctype, bom, tuple(keys), tuple(columns))
+
+	for row in repeated:
+		line = representative.get(tuple(row.get(key) for key in keys))
+		if not line:
+			continue
+		for column in columns:
+			row[column] = line.get(column)
+
+
+@request_cache
+def _representative_lines(doctype, bom, keys, columns):
+	"""Cached per request: get_bom_items_as_dict recurses through phantom BOMs, and the same
+	sub-BOM is commonly reached more than once."""
+	representative = {}
+	for line in frappe.get_all(
+		doctype,
+		filters={"parent": bom, "parenttype": "BOM", "docstatus": ("<", 2)},
+		fields=[*keys, *columns],
+		order_by="idx",
+	):
+		representative.setdefault(tuple(line.get(key) for key in keys), line)
+
+	return representative
 
 
 def _get_bom_item_tables(opts):
@@ -1264,16 +1323,16 @@ def _build_base_bom_items_query(bom, company, qty, t):
 def _add_bom_item_columns(query, t, bom, opts, track_semi_finished_goods):
 	is_stock_item = cint(not opts.include_non_stock_items)
 	stock_item_condition = t.item_doc.is_stock_item.isin([1, is_stock_item])
-	# rate is constant per grouped item -> Max() keeps it out of the Sum (preserving the original
-	# Sum(...) * rate * qty arithmetic) while making the expression postgres-valid under GROUP BY.
-	amount_col = (
-		Sum(t.bom_item.stock_qty / IfNull(t.bom_doc.quantity, 1)) * Max(t.bom_item.rate) * opts.qty
-	).as_("amount")
+	if opts.fetch_secondary_items:
+		return _add_secondary_item_columns(query, t, stock_item_condition)
+
+	# BOM Item rate is per row UOM, while BOM Explosion Item rate is per stock UOM. Select the
+	# matching quantity so a normal BOM row's conversion factor is not applied twice.
+	qty_col = t.bom_item.stock_qty if cint(opts.fetch_exploded) else t.bom_item.qty
+	amount_col = (Sum(qty_col / IfNull(t.bom_doc.quantity, 1) * t.bom_item.rate) * opts.qty).as_("amount")
 
 	if cint(opts.fetch_exploded):
 		return _add_exploded_item_columns(query, t, bom, amount_col, stock_item_condition)
-	if opts.fetch_secondary_items:
-		return _add_secondary_item_columns(query, t, stock_item_condition)
 	return _add_normal_item_columns(query, t, amount_col, stock_item_condition, track_semi_finished_goods)
 
 
@@ -1290,10 +1349,11 @@ def _add_exploded_item_columns(query, t, bom, amount_col, stock_item_condition):
 	# keeping the GROUP BY postgres-valid; the correlated idx subquery references only item_code
 	# (a grouped column) so it stays valid and still overrides the explosion idx for display.
 	query = query.select(
+		Max(t.bom_item.description).as_("description"),
 		Max(t.bom_item.source_warehouse).as_("source_warehouse"),
+		Count(t.bom_item.name).distinct().as_("line_count"),
 		Max(t.bom_item.operation).as_("operation"),
 		Max(t.bom_item.include_item_in_manufacturing).as_("include_item_in_manufacturing"),
-		Max(t.bom_item.description).as_("description"),
 		Max(t.bom_item.rate).as_("rate"),
 		Max(t.bom_item.sourced_by_supplier).as_("sourced_by_supplier"),
 		amount_col,
@@ -1329,14 +1389,15 @@ def _add_normal_item_columns(query, t, amount_col, stock_item_condition, track_s
 	# under the same alias and silently shadowed (last value wins in the dict), so it is dropped here
 	# -- output is unchanged.
 	query = query.select(
-		Max(t.bom_item.uom).as_("uom"),
-		Max(t.bom_item.conversion_factor).as_("conversion_factor"),
+		Max(t.bom_item.description).as_("description"),
 		Max(t.bom_item.source_warehouse).as_("source_warehouse"),
+		Count(t.bom_item.name).distinct().as_("line_count"),
 		Max(t.bom_item.operation).as_("operation"),
 		Max(t.bom_item.include_item_in_manufacturing).as_("include_item_in_manufacturing"),
 		Max(t.bom_item.sourced_by_supplier).as_("sourced_by_supplier"),
+		Max(t.bom_item.uom).as_("uom"),
+		Max(t.bom_item.conversion_factor).as_("conversion_factor"),
 		amount_col,
-		Max(t.bom_item.description).as_("description"),
 		Max(t.bom_item.base_rate).as_("rate"),
 		Max(t.bom_item.operation_row_id).as_("operation_row_id"),
 		t.bom_item.is_phantom_item,

erpnext/manufacturing/doctype/bom/test_bom.py

@@ -101,6 +101,70 @@ class TestBOM(ERPNextTestSuite):
 			self.assertEqual(flt(items_dict[component].qty), 1.0)
 			self.assertNotIn(rm_normal, items_dict)
 
+	@timeout
+	def test_get_items_amount_uses_each_lines_own_rate(self):
+		from erpnext.manufacturing.doctype.bom.bom import get_bom_items_as_dict
+		from erpnext.manufacturing.doctype.production_plan.test_production_plan import make_bom
+
+		rm = make_item(properties={"is_stock_item": 1, "valuation_rate": 10, "stock_uom": "Nos"})
+		if not any(row.uom == "Box" for row in rm.uoms):
+			rm.append("uoms", {"uom": "Box", "conversion_factor": 5})
+			rm.save()
+
+		fg_item = make_item(properties={"is_stock_item": 1, "valuation_rate": 10}).name
+		bom = make_bom(item=fg_item, raw_materials=[rm.name], rm_qty=2, do_not_save=True)
+		bom.append("items", {"item_code": rm.name, "qty": 3, "uom": "Box", "stock_uom": "Nos"})
+		bom.save()
+		bom.submit()
+
+		lines = [row for row in bom.items if row.item_code == rm.name]
+		self.assertEqual(len(lines), 2)
+		self.assertEqual(len({flt(row.rate) for row in lines}), 2)
+
+		requested_qty = 2
+		expected = sum(flt(row.qty) * flt(row.rate) for row in lines) / flt(bom.quantity) * requested_qty
+		items_dict = get_bom_items_as_dict(bom.name, "_Test Company", qty=requested_qty, fetch_exploded=0)
+
+		self.assertEqual(len([row for row in items_dict if row == rm.name]), 1)
+		self.assertAlmostEqual(flt(items_dict[rm.name].amount), expected, places=2)
+
+	@timeout
+	def test_get_items_takes_line_columns_from_one_line(self):
+		from erpnext.manufacturing.doctype.bom.bom import get_bom_items_as_dict
+		from erpnext.manufacturing.doctype.production_plan.test_production_plan import make_bom
+		from erpnext.stock.doctype.warehouse.test_warehouse import create_warehouse
+
+		rm = make_item(properties={"is_stock_item": 1, "valuation_rate": 10})
+		fg_item = make_item(properties={"is_stock_item": 1, "valuation_rate": 10}).name
+
+		first_warehouse = create_warehouse("_Test BOM Line A")
+		second_warehouse = create_warehouse("_Test BOM Line B")
+
+		bom = make_bom(item=fg_item, raw_materials=[rm.name], rm_qty=2, do_not_save=True)
+		bom.items[0].description = "bbb first line"
+		bom.items[0].source_warehouse = first_warehouse
+		bom.append(
+			"items",
+			{
+				"item_code": rm.name,
+				"qty": 3,
+				"uom": rm.stock_uom,
+				"stock_uom": rm.stock_uom,
+				"description": "ccc second line",
+				"source_warehouse": second_warehouse,
+			},
+		)
+		bom.save()
+		bom.submit()
+
+		items_dict = get_bom_items_as_dict(bom.name, "_Test Company", qty=1, fetch_exploded=0)
+		row = items_dict[rm.name]
+
+		# "ccc" sorts above "bbb" on either engine, so an aggregated description would win here;
+		# the value must instead come from the first line, together with that line's warehouse
+		self.assertEqual(row.description, "bbb first line")
+		self.assertEqual(row.source_warehouse, first_warehouse)
+
 	@timeout
 	def test_default_bom(self):
 		def _get_default_bom_in_item():

erpnext/manufacturing/doctype/plant_floor/stock_summary_template.html

@@ -2,9 +2,9 @@
 <div class="row" style="border-bottom:1px solid var(--border-color); padding:4px 5px; margin-top: 3px;margin-bottom: 3px;">
 	<div class="col-sm-1">
 		{% if(row.image) { %}
-			<img style="width:50px;height:50px;" src="{{row.image}}">
+			<img style="width:50px;height:50px;" src="{{frappe.utils.escape_html(row.image)}}">
 		{% } else { %}
-			<div style="width:50px;height:50px;background-color:var(--control-bg);text-align:center;padding-top:15px">{{frappe.get_abbr(row.item_code, 2)}}</div>
+			<div style="width:50px;height:50px;background-color:var(--control-bg);text-align:center;padding-top:15px">{{frappe.get_abbr(frappe.utils.escape_html(row.item_code), 2)}}</div>
 		{% } %}
 	</div>
 	<div class="col-sm-3">
@@ -13,7 +13,7 @@
 		{% } else { %}
 			{{row.item_link}}
 			<p>
-				{{row.item_name}}
+				{{frappe.utils.escape_html(row.item_name)}}
 			</p>
 		{% } %}
 
@@ -52,10 +52,10 @@
 		</span>
 	</div>
 	<div class="col-sm-1">
-		<button style="margin-left: 7px;" class="btn btn-default btn-xs btn-add" data-item-code="{{ escape(row.item_code) }}">{{ __("Add") }}</button>
+		<button style="margin-left: 7px;" class="btn btn-default btn-xs btn-add" data-item-code="{{ frappe.utils.escape_html(row.item_code) }}">{{ __("Add") }}</button>
 	</div>
 	<div class="col-sm-1">
-		<button style="margin-left: 7px;" class="btn btn-default btn-xs btn-move" data-item-code="{{ escape(row.item_code) }}">{{ __("Move") }}</button>
+		<button style="margin-left: 7px;" class="btn btn-default btn-xs btn-move" data-item-code="{{ frappe.utils.escape_html(row.item_code) }}">{{ __("Move") }}</button>
 	</div>
 </div>
 {% }); %}

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

@@ -4,7 +4,8 @@
 """BOM explosion helpers for Production Plan material planning."""
 
 import frappe
-from frappe.query_builder.functions import IfNull, Max, Min, Sum
+from frappe.query_builder.functions import Count, IfNull, Max, Min, Sum
+from frappe.utils.caching import request_cache
 
 from erpnext.manufacturing.doctype.production_plan.services.planning_queries import get_uom_conversion_factor
 
@@ -21,7 +22,7 @@ def _exploded_items_query(company, bom_no, include_non_stock_items, planned_qty)
 	item = frappe.qb.DocType("Item")
 	item_default = frappe.qb.DocType("Item Default")
 	item_uom = frappe.qb.DocType("UOM Conversion Detail")
-	return (
+	rows = (
 		frappe.qb.from_(bei)
 		.join(bom)
 		.on(bom.name == bei.parent)
@@ -36,19 +37,92 @@ def _exploded_items_query(company, bom_no, include_non_stock_items, planned_qty)
 		.groupby(bei.item_code, bei.stock_uom)
 	).run(as_dict=True)
 
+	_apply_representative_lines(
+		rows, "BOM Explosion Item", bom_no, ("item_code", "stock_uom"), include_non_stock_items
+	)
+	return rows
+
+
+def _apply_representative_lines(rows, doctype, bom_no, keys, include_non_stock_items=True):
+	"""Fill description/source_warehouse from a single real BOM line per group.
+
+	Both describe a line, not an item, so a BOM listing the same item more than once holds
+	several values per group. Aggregating each independently can pair one line's description
+	with another's warehouse, and Max() over text is a sort -- MariaDB folds case, PostgreSQL
+	orders by byte value, so the two engines pick differently. Take the first line by idx.
+
+	Only groups built from more than one line need this. Where a group has a single line, Max() of
+	one value is that value, so the selected columns are already exact and no query is issued --
+	which matters because this runs once per BOM in a recursive explosion.
+	"""
+	repeated = [row for row in rows if (row.pop("line_count", 1) or 1) > 1]
+	if not repeated:
+		return
+
+	representative = _representative_lines(doctype, bom_no, tuple(keys), include_non_stock_items)
+
+	for row in repeated:
+		line = representative.get(tuple(row.get(key) for key in keys))
+		if line:
+			row.description = line.description
+			row.source_warehouse = line.source_warehouse
+
+
+@request_cache
+def _representative_lines(doctype, bom_no, keys, include_non_stock_items):
+	"""Cached per request: the explosion recurses and commonly revisits the same sub-BOM."""
+	# only BOM Item carries is_phantom_item, and only its query ORs the phantom flag into the stock
+	# filter; the explosion table has neither
+	filters_phantom = doctype == "BOM Item"
+	fields = ["item_code", "stock_uom", "description", "source_warehouse"]
+	if filters_phantom:
+		fields.append("is_phantom_item")
+
+	lines = frappe.get_all(
+		doctype,
+		filters={
+			"parent": bom_no,
+			"parenttype": "BOM",
+			"is_sub_assembly_item": 0,
+			"docstatus": ("<", 2),
+		},
+		fields=fields,
+		order_by="idx",
+	)
+
+	# mirror the caller's stock filter: a non-stock line the main query excluded must not become
+	# the representative for a group that only exists because of a phantom line
+	if not include_non_stock_items and filters_phantom and lines:
+		stock_items = set(
+			frappe.get_all(
+				"Item",
+				filters={"name": ("in", list({line.item_code for line in lines})), "is_stock_item": 1},
+				pluck="name",
+			)
+		)
+		lines = [line for line in lines if line.item_code in stock_items or line.is_phantom_item]
+
+	representative = {}
+	for line in lines:
+		representative.setdefault(tuple(line.get(key) for key in keys), line)
+
+	return representative
+
 
 def _exploded_item_columns(bei, bom, item, item_default, item_uom, planned_qty):
-	# only item_code/stock_uom are grouped; the rest are functionally dependent on the grouped item
-	# or arbitrary per BOM Item on MySQL -> Max() keeps the GROUP BY valid on postgres with the same
-	# value MySQL picked.
+	# every column here is functionally dependent on the grouped item_code -- Item, Item Default and
+	# UOM Conversion Detail are joined on it and the BOM is pinned by the filter -- so Max() returns
+	# their single value. The BOM-line columns come from a representative line instead; see
+	# _apply_representative_lines.
 	return [
 		(IfNull(Sum(bei.stock_qty / IfNull(bom.quantity, 1)), 0) * planned_qty).as_("qty"),
 		Max(item.item_name).as_("item_name"),
 		Max(item.name).as_("item_code"),
 		Max(bei.description).as_("description"),
+		Max(bei.source_warehouse).as_("source_warehouse"),
+		Count(bei.name).distinct().as_("line_count"),
 		bei.stock_uom,
 		Max(item.min_order_qty).as_("min_order_qty"),
-		Max(bei.source_warehouse).as_("source_warehouse"),
 		Max(item.default_material_request_type).as_("default_material_request_type"),
 		Max(item.min_order_qty).as_("min_order_qty"),
 		Max(item_default.default_warehouse).as_("default_warehouse"),
@@ -96,7 +170,7 @@ def _subitems_query(company, bom_no, include_non_stock_items, parent_qty, planne
 	item = frappe.qb.DocType("Item")
 	item_default = frappe.qb.DocType("Item Default")
 	item_uom = frappe.qb.DocType("UOM Conversion Detail")
-	return (
+	rows = (
 		frappe.qb.from_(bom_item)
 		.join(bom)
 		.on(bom.name == bom_item.parent)
@@ -113,6 +187,9 @@ def _subitems_query(company, bom_no, include_non_stock_items, parent_qty, planne
 		.orderby(Min(bom_item.idx))
 	).run(as_dict=True)
 
+	_apply_representative_lines(rows, "BOM Item", bom_no, ("item_code",), include_non_stock_items)
+	return rows
+
 
 def _subitem_columns(bom_item, bom, item, item_default, item_uom, parent_qty, planned_qty):
 	qty = IfNull(parent_qty * Sum(bom_item.stock_qty / IfNull(bom.quantity, 1)) * planned_qty, 0).as_("qty")
@@ -128,9 +205,10 @@ def _subitem_columns(bom_item, bom, item, item_default, item_uom, parent_qty, pl
 		Max(item.item_name).as_("item_name"),
 		qty,
 		Max(item.is_sub_contracted_item).as_("is_sub_contracted"),
-		Max(bom_item.source_warehouse).as_("source_warehouse"),
-		Max(item.default_bom).as_("default_bom"),
 		Max(bom_item.description).as_("description"),
+		Max(bom_item.source_warehouse).as_("source_warehouse"),
+		Count(bom_item.name).distinct().as_("line_count"),
+		Max(item.default_bom).as_("default_bom"),
 		Max(bom_item.stock_uom).as_("stock_uom"),
 		Max(item.min_order_qty).as_("min_order_qty"),
 		Max(item.safety_stock).as_("safety_stock"),

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

@@ -4,8 +4,9 @@
 """Sub-assembly resolution helpers for Production Plan."""
 
 import frappe
-from frappe.query_builder.functions import IfNull, Max, Sum
+from frappe.query_builder.functions import Count, IfNull, Max, Sum
 from frappe.utils import flt
+from frappe.utils.caching import request_cache
 
 from erpnext.manufacturing.doctype.bom.bom import get_children as get_bom_children
 from erpnext.manufacturing.doctype.production_plan.services.planning_queries import (
@@ -167,7 +168,7 @@ def _sub_assembly_rm_query(company, bom_no, include_non_stock_items, planned_qty
 	item = frappe.qb.DocType("Item")
 	item_default = frappe.qb.DocType("Item Default")
 	item_uom = frappe.qb.DocType("UOM Conversion Detail")
-	return (
+	rows = (
 		frappe.qb.from_(bei)
 		.join(bom)
 		.on(bom.name == bei.parent)
@@ -182,6 +183,50 @@ def _sub_assembly_rm_query(company, bom_no, include_non_stock_items, planned_qty
 		.groupby(bei.item_code, bei.stock_uom, bei.bom_no, bei.is_phantom_item)
 	).run(as_dict=True)
 
+	_apply_representative_lines(rows, bom_no)
+	return rows
+
+
+def _apply_representative_lines(rows, bom_no):
+	"""Fill description/source_warehouse from a single real BOM Item line per group.
+
+	Both describe a line, not an item. Aggregating each independently can pair one line's
+	description with another's warehouse, and Max() over text is a sort -- MariaDB folds case,
+	PostgreSQL orders by byte value, so the engines pick differently. Take the first line by idx.
+	"""
+	repeated = [row for row in rows if (row.pop("line_count", 1) or 1) > 1]
+	if not repeated:
+		return
+
+	keys = ("item_code", "stock_uom", "bom_no", "is_phantom_item")
+	representative = _representative_lines(bom_no, keys)
+
+	for row in repeated:
+		line = representative.get(tuple(row.get(key) for key in keys))
+		if line:
+			row.description = line.description
+			row.source_warehouse = line.source_warehouse
+
+
+@request_cache
+def _representative_lines(bom_no, keys):
+	"""Cached per request: sub-assembly resolution recurses and revisits the same BOM."""
+	representative = {}
+	for line in frappe.get_all(
+		"BOM Item",
+		filters={
+			"parent": bom_no,
+			"parenttype": "BOM",
+			"is_sub_assembly_item": 0,
+			"docstatus": 1,
+		},
+		fields=["item_code", "stock_uom", "bom_no", "is_phantom_item", "description", "source_warehouse"],
+		order_by="idx",
+	):
+		representative.setdefault(tuple(line.get(key) for key in keys), line)
+
+	return representative
+
 
 def _sub_assembly_rm_columns(bei, bom, item, item_default, item_uom, planned_qty):
 	# Grouped by item_code/stock_uom plus bom_no/is_phantom_item: those two MUST come from the same
@@ -195,11 +240,12 @@ def _sub_assembly_rm_columns(bei, bom, item, item_default, item_uom, planned_qty
 		Max(item.item_name).as_("item_name"),
 		Max(item.name).as_("item_code"),
 		Max(bei.description).as_("description"),
+		Max(bei.source_warehouse).as_("source_warehouse"),
+		Count(bei.name).distinct().as_("line_count"),
 		bei.stock_uom,
 		bei.is_phantom_item,
 		bei.bom_no,
 		Max(item.min_order_qty).as_("min_order_qty"),
-		Max(bei.source_warehouse).as_("source_warehouse"),
 		Max(item.default_material_request_type).as_("default_material_request_type"),
 		Max(item.min_order_qty).as_("min_order_qty"),
 		Max(item_default.default_warehouse).as_("default_warehouse"),

erpnext/manufacturing/doctype/work_order/test_work_order.py

@@ -2889,6 +2889,99 @@ class TestWorkOrder(ERPNextTestSuite):
 			f"BOM-path disassembly must apply process_loss_per; expected 18, got {bom_scrap_row.qty}",
 		)
 
+	def test_disassembly_mixed_uom_rows_are_aggregated_in_stock_uom(self):
+		"""Quantities and rates from different row UOMs must be aggregated in stock UOM."""
+		from erpnext.stock.doctype.stock_entry.services.disassemble import DisassembleStockEntry
+		from erpnext.stock.doctype.stock_entry.test_stock_entry import (
+			make_stock_entry as make_stock_entry_test_record,
+		)
+
+		raw_item_doc = make_item(
+			"Test Raw for Disassembly Coherence", {"is_stock_item": 1, "stock_uom": "Nos"}
+		)
+		box_uom = next((row for row in raw_item_doc.uoms if row.uom == "Box"), None)
+		if box_uom:
+			box_uom.conversion_factor = 5
+		else:
+			raw_item_doc.append("uoms", {"uom": "Box", "conversion_factor": 5})
+		raw_item_doc.save()
+		raw_item = raw_item_doc.name
+		fg_item = make_item("Test FG for Disassembly Coherence", {"is_stock_item": 1}).name
+		bom = make_bom(item=fg_item, quantity=1, raw_materials=[raw_item], rm_qty=2)
+
+		wo = make_wo_order_test_record(production_item=fg_item, qty=10, bom_no=bom.name, status="Not Started")
+		make_stock_entry_test_record(
+			item_code=raw_item,
+			purpose="Material Receipt",
+			target=wo.wip_warehouse,
+			qty=50,
+			basic_rate=100,
+		)
+
+		transfer = frappe.get_doc(make_stock_entry(wo.name, "Material Transfer for Manufacture", wo.qty))
+		for item in transfer.items:
+			item.s_warehouse = wo.wip_warehouse
+		transfer.save()
+		transfer.submit()
+
+		first = frappe.get_doc(make_stock_entry(wo.name, "Manufacture", 5))
+		first.submit()
+		second = frappe.get_doc(make_stock_entry(wo.name, "Manufacture", 5))
+		second.submit()
+		wo.reload()
+
+		first_row = next(row for row in first.items if row.item_code == raw_item)
+		second_row = next(row for row in second.items if row.item_code == raw_item)
+		first_stock_qty = flt(first_row.transfer_qty)
+		frappe.db.set_value(
+			"Stock Entry Detail",
+			first_row.name,
+			{
+				"uom": "Box",
+				"conversion_factor": 5,
+				"qty": first_stock_qty / 5,
+				"transfer_qty": first_stock_qty,
+				"basic_rate": 100,
+			},
+			update_modified=False,
+		)
+		frappe.db.set_value("Stock Entry Detail", second_row.name, "basic_rate", 200, update_modified=False)
+
+		posted_rows = frappe.get_all(
+			"Stock Entry Detail",
+			filters={"parent": ("in", [first.name, second.name]), "item_code": raw_item},
+			fields=["qty", "transfer_qty", "uom", "conversion_factor", "basic_rate"],
+		)
+		self.assertEqual(len({row.uom for row in posted_rows}), 2)
+		self.assertTrue(
+			all(flt(row.qty) * flt(row.conversion_factor) == flt(row.transfer_qty) for row in posted_rows)
+		)
+
+		service = DisassembleStockEntry(frappe._dict(work_order=wo.name, source_stock_entry=None))
+		source_row = next(
+			row for row in service.get_items_from_manufacture_stock_entry() if row.item_code == raw_item
+		)
+
+		expected_stock_qty = sum(flt(row.transfer_qty) for row in posted_rows)
+		expected_rate = (
+			sum(flt(row.basic_rate) * flt(row.transfer_qty) for row in posted_rows) / expected_stock_qty
+		)
+		self.assertEqual(source_row.uom, source_row.stock_uom)
+		self.assertEqual(flt(source_row.conversion_factor), 1.0)
+		self.assertEqual(flt(source_row.qty), expected_stock_qty)
+		self.assertEqual(flt(source_row.transfer_qty), expected_stock_qty)
+		self.assertAlmostEqual(flt(source_row.basic_rate), expected_rate, places=6)
+
+		disassemble_qty = 4
+		disassembly = frappe.get_doc(make_stock_entry(wo.name, "Disassemble", disassemble_qty))
+		disassembly.save()
+		disassembly_row = next(row for row in disassembly.items if row.item_code == raw_item)
+		expected_disassembly_qty = expected_stock_qty * disassemble_qty / flt(wo.produced_qty)
+		self.assertEqual(disassembly_row.uom, disassembly_row.stock_uom)
+		self.assertEqual(flt(disassembly_row.conversion_factor), 1.0)
+		self.assertEqual(flt(disassembly_row.transfer_qty), expected_disassembly_qty)
+		disassembly.submit()
+
 	def test_disassembly_with_additional_rm_not_in_bom(self):
 		"""
 		Test that SE-linked disassembly includes additional raw materials

erpnext/manufacturing/doctype/workstation/workstation.py

@@ -457,7 +457,7 @@ def get_workstations(**kwargs):
 		d.color = color_map.get(d.status, "red")
 		d.workstation_link = get_url_to_form("Workstation", d.name)
 		if d.status != "Production":
-			d.status_image = d.off_status_image
+			d.status_image = frappe.utils.escape_html(d.off_status_image)
 			d.workstation_off = "workstation-off"
 
 	return data

erpnext/manufacturing/report/bom_stock_analysis/bom_stock_analysis.py

@@ -134,15 +134,15 @@ def get_data_without_qty_to_make(filters):
 	for row in raw_rows:
 		data.append(
 			{
-				"item": row[0],
-				"description": row[1],
-				"from_bom_no": row[2],
-				"qty_per_unit": fmt_qty(row[3]),
-				"available_qty": fmt_qty(row[4]),
+				"item": row.item_code,
+				"description": row.description,
+				"from_bom_no": row.from_bom_no,
+				"qty_per_unit": fmt_qty(row.qty_per_unit),
+				"available_qty": fmt_qty(row.available_qty),
 			}
 		)
 
-	min_producible = min((row[5] or 0) for row in raw_rows) if raw_rows else 0
+	min_producible = min((row.producible_qty or 0) for row in raw_rows) if raw_rows else 0
 	# blank spacer row
 	data.append({})
 
@@ -190,27 +190,14 @@ def batch_fetch_purchase_rates(bom_data):
 	}
 
 
-def get_bom_data(filters):
-	bom_item_table = "BOM Explosion Item" if filters.get("show_exploded_view") else "BOM Item"
-
-	bom_item = frappe.qb.DocType(bom_item_table)
+def get_stock_qty_by_item(filters):
+	"""One row per item_code, so joining it to BOM Item cannot multiply either side's sum."""
 	bin = frappe.qb.DocType("Bin")
 
 	query = (
-		frappe.qb.from_(bom_item)
-		.left_join(bin)
-		.on(bom_item.item_code == bin.item_code)
-		.select(
-			bom_item.item_code,
-			# non-grouped columns are constant per grouped item_code -> Max() keeps the GROUP BY valid
-			Max(bom_item.description).as_("description"),
-			Max(bom_item.parent).as_("from_bom_no"),
-			Sum(bom_item.qty_consumed_per_unit).as_("qty_per_unit"),
-			IfNull(Sum(bin.actual_qty), 0).as_("actual_qty"),
-		)
-		.where((bom_item.parent == filters.get("bom")) & (bom_item.parenttype == "BOM"))
-		.groupby(bom_item.item_code)
-		.orderby(Min(bom_item.idx))
+		frappe.qb.from_(bin)
+		.select(bin.item_code, Sum(bin.actual_qty).as_("actual_qty"))
+		.groupby(bin.item_code)
 	)
 
 	if filters.get("warehouse"):
@@ -233,30 +220,64 @@ def get_bom_data(filters):
 		else:
 			query = query.where(bin.warehouse == filters.get("warehouse"))
 
+	return query
+
+
+def get_bom_data(filters):
+	bom_item_table = "BOM Explosion Item" if filters.get("show_exploded_view") else "BOM Item"
+
+	bom_item = frappe.qb.DocType(bom_item_table)
+	stock_qty = get_stock_qty_by_item(filters).as_("stock_qty")
+
+	base = frappe.qb.from_(bom_item)
+	base = base.join(stock_qty) if filters.get("warehouse") else base.left_join(stock_qty)
+
+	query = (
+		base.on(bom_item.item_code == stock_qty.item_code)
+		.select(
+			bom_item.item_code,
+			# non-grouped columns are constant per grouped item_code -> Max() keeps the GROUP BY valid
+			Max(bom_item.parent).as_("from_bom_no"),
+			Sum(bom_item.qty_consumed_per_unit).as_("qty_per_unit"),
+			IfNull(Max(stock_qty.actual_qty), 0).as_("actual_qty"),
+		)
+		.where((bom_item.parent == filters.get("bom")) & (bom_item.parenttype == "BOM"))
+		.groupby(bom_item.item_code)
+		.orderby(Min(bom_item.idx))
+	)
+
 	data = query.run(as_dict=True)
 
+	# description belongs to a BOM line, not to the item, so a component listed more than once holds
+	# several values per group. Max() over text is a sort and the engines sort text differently
+	# (MariaDB folds case, PostgreSQL orders by byte value), so read it off one real line instead.
+	# For BOM Item that same line also supplies bom_no + is_phantom_item, which drive whether and
+	# which sub-BOM explode_phantom_boms recurses into and so must stay coherent with each other:
+	# the first line, upgraded to the first phantom line if any exists, so a phantom sub-BOM is never
+	# dropped just because a non-phantom line happens to be listed first.
+	fields = ["item_code", "description"]
+	if bom_item_table == "BOM Item":
+		fields += ["bom_no", "is_phantom_item"]
+
+	representative = {}
+	for line in frappe.get_all(
+		bom_item_table,
+		filters={"parent": filters.get("bom"), "parenttype": "BOM"},
+		fields=fields,
+		order_by="idx",
+	):
+		existing = representative.get(line.item_code)
+		if existing is None or (line.get("is_phantom_item") and not existing.get("is_phantom_item")):
+			representative[line.item_code] = line
+
+	for row in data:
+		line = representative.get(row.item_code)
+		row.description = line.description if line else None
+		if bom_item_table == "BOM Item":
+			row.bom_no = line.bom_no if line else None
+			row.is_phantom_item = line.is_phantom_item if line else None
+
 	if bom_item_table == "BOM Item":
-		# bom_no + is_phantom_item drive whether/which sub-BOM explode_phantom_boms recurses into, so
-		# they must come from the SAME BOM Item line. Aggregating each independently (Max) could pair a
-		# bom_no from one line with is_phantom_item from another when an item_code repeats in the BOM.
-		# Rows are grouped by item_code (one qty_per_unit total per component), so pick one coherent
-		# representative line: the first line, but upgrade to the first phantom line if any exists, so a
-		# phantom sub-BOM is never dropped just because a non-phantom line happens to be listed first.
-		representative = {}
-		for line in frappe.get_all(
-			"BOM Item",
-			filters={"parent": filters.get("bom"), "parenttype": "BOM"},
-			fields=["item_code", "bom_no", "is_phantom_item"],
-			order_by="idx",
-		):
-			existing = representative.get(line.item_code)
-			if existing is None or (line.is_phantom_item and not existing.is_phantom_item):
-				representative[line.item_code] = line
-		for row in data:
-			line = representative.get(row.item_code)
-			if line:
-				row.bom_no = line.bom_no
-				row.is_phantom_item = line.is_phantom_item
 		return explode_phantom_boms(data, filters)
 
 	return data
@@ -337,15 +358,37 @@ def get_producible_fg_items(filters):
 			BOM_ITEM.item_code,
 			# Sum() below makes this an aggregate query; the other columns are constant per grouped
 			# item_code -> Max() keeps them valid on postgres with the same value MySQL picked.
-			Max(BOM_ITEM.description).as_("description"),
+			# description is not: it belongs to the line, so it comes from a representative one below.
 			Max(BOM_ITEM.parent).as_("from_bom_no"),
 			Max(BOM_ITEM.stock_qty / BOM.quantity).as_("qty_per_unit"),
 			Max(IfNull(bin_subquery.actual_qty, 0)).as_("available_qty"),
-			Floor(Max(bin_subquery.actual_qty) / ((Sum(BOM_ITEM.stock_qty)) / Max(BOM.quantity))),
+			Floor(Max(bin_subquery.actual_qty) / ((Sum(BOM_ITEM.stock_qty)) / Max(BOM.quantity))).as_(
+				"producible_qty"
+			),
 		)
 		.where((BOM_ITEM.parent == filters.get("bom")) & (BOM_ITEM.parenttype == "BOM"))
 		.groupby(BOM_ITEM.item_code)
 		.orderby(Min(BOM_ITEM.idx))
 	)
 
-	return query.run(as_list=True)
+	rows = query.run(as_dict=True)
+	descriptions = get_representative_descriptions("BOM Item", filters.get("bom"))
+	for row in rows:
+		row.description = descriptions.get(row.item_code)
+
+	return rows
+
+
+def get_representative_descriptions(doctype, bom):
+	"""First line by idx per item_code. description belongs to a line, not an item, so aggregating it
+	sorts text -- and MariaDB folds case while PostgreSQL orders by byte value."""
+	descriptions = {}
+	for line in frappe.get_all(
+		doctype,
+		filters={"parent": bom, "parenttype": "BOM"},
+		fields=["item_code", "description"],
+		order_by="idx",
+	):
+		descriptions.setdefault(line.item_code, line.description)
+
+	return descriptions

erpnext/manufacturing/report/bom_stock_analysis/test_bom_stock_analysis.py

@@ -1,13 +1,18 @@
 # Copyright (c) 2026, Frappe Technologies Pvt. Ltd. and contributors
 # For license information, please see license.txt
 import frappe
-from frappe.utils import fmt_money
+from frappe.utils import flt, fmt_money
 
 from erpnext.manufacturing.doctype.production_plan.test_production_plan import make_bom
 from erpnext.manufacturing.report.bom_stock_analysis.bom_stock_analysis import (
 	execute as bom_stock_analysis_report,
 )
+from erpnext.manufacturing.report.bom_stock_analysis.bom_stock_analysis import get_bom_data
 from erpnext.stock.doctype.item.test_item import make_item
+from erpnext.stock.doctype.stock_reconciliation.test_stock_reconciliation import (
+	create_stock_reconciliation,
+)
+from erpnext.stock.doctype.warehouse.test_warehouse import create_warehouse
 from erpnext.tests.utils import ERPNextTestSuite
 
 
@@ -146,6 +151,41 @@ class TestBOMStockAnalysis(ERPNextTestSuite):
 		"""
 		self._assert_phantom_exploded(*self._build_duplicate_component_bom(phantom_first=False))
 
+	def test_bom_data_is_not_multiplied_by_the_bin_join(self):
+		"""Bin joins one row per warehouse, BOM Item one per line -- neither sum may count the other.
+
+		With the component listed on two BOM lines and stocked in two warehouses, the join yields
+		four rows. Summing qty_consumed_per_unit over it counts each line once per warehouse, and
+		summing actual_qty counts each warehouse once per line.
+		"""
+		rm = make_item(properties={"is_stock_item": 1, "valuation_rate": 10})
+		fg = make_item(properties={"is_stock_item": 1, "valuation_rate": 10}).name
+
+		bom = make_bom(item=fg, raw_materials=[rm.name], rm_qty=2, do_not_save=True)
+		bom.append(
+			"items",
+			{"item_code": rm.name, "qty": 3, "uom": rm.stock_uom, "stock_uom": rm.stock_uom},
+		)
+		bom.save()
+		bom.submit()
+
+		for suffix, qty in (("A", 6), ("B", 4)):
+			warehouse = create_warehouse(f"_Test BOM Stock Analysis {suffix}")
+			create_stock_reconciliation(item_code=rm.name, warehouse=warehouse, qty=qty, rate=10)
+
+		rows = [row for row in get_bom_data({"bom": bom.name}) if row.item_code == rm.name]
+		self.assertEqual(len(rows), 1)
+
+		lines = [line for line in bom.items if line.item_code == rm.name]
+		self.assertEqual(len(lines), 2)
+
+		self.assertAlmostEqual(
+			flt(rows[0].qty_per_unit),
+			sum(flt(line.qty_consumed_per_unit) for line in lines),
+			places=6,
+		)
+		self.assertAlmostEqual(flt(rows[0].actual_qty), 10.0, places=6)
+
 
 def split_data_and_footer(raw_data):
 	"""Separate component rows from the footer row. Skips blank spacer rows."""

erpnext/public/js/bom_configurator/bom_configurator.bundle.js

@@ -32,18 +32,7 @@ class BOMConfigurator {
 	}
 
 	bind_events() {
-		frappe.views.trees["BOM Configurator"].events = {
-			frm: this.frm,
-			add_item: this.add_item,
-			add_sub_assembly: this.add_sub_assembly,
-			set_query_for_workstation: this.set_query_for_workstation,
-			get_sub_assembly_modal_fields: this.get_sub_assembly_modal_fields,
-			convert_to_sub_assembly: this.convert_to_sub_assembly,
-			delete_node: this.delete_node,
-			edit_bom: this.edit_bom,
-			load_tree: this.load_tree,
-			set_default_qty: this.set_default_qty,
-		};
+		frappe.views.trees["BOM Configurator"].events = this;
 	}
 
 	tree_options() {

erpnext/public/js/controllers/transaction.js

@@ -18,10 +18,15 @@ erpnext.stock.qi_outgoing_purposes = [
 	"Subcontracting Delivery",
 	"Disassemble",
 ];
+erpnext.stock.secondary_item_purposes = ["Manufacture", "Repack", "Disassemble"];
 erpnext.stock.is_incoming_qi_purpose = (purpose) =>
 	purpose === "Manufacture" || erpnext.stock.qi_incoming_purposes.includes(purpose);
 erpnext.stock.row_requires_quality_inspection = (purpose, row) => {
-	if (row.secondary_item_type || row.is_legacy_scrap_item) return false;
+	if (
+		erpnext.stock.secondary_item_purposes.includes(purpose) &&
+		(row.secondary_item_type || row.is_legacy_scrap_item)
+	)
+		return false;
 	if (purpose === "Manufacture") return !!row.is_finished_item;
 	if (erpnext.stock.qi_incoming_purposes.includes(purpose)) return !!row.t_warehouse;
 	if (erpnext.stock.qi_outgoing_purposes.includes(purpose))

erpnext/public/js/plant_floor_visual/visual_plant.js

@@ -176,7 +176,7 @@ class VisualPlantFloor {
 				.find(".workstation-image-container")
 				.append(
 					`<div class="workstation-image-cls workstation-abbr" style="margin:6px; height:82px">${frappe.get_abbr(
-						data.name,
+						frappe.utils.escape_html(data.name),
 						2
 					)}</div>`
 				);

erpnext/public/js/templates/item_selector.html

@@ -1,17 +1,19 @@
 <div class="app-listing item-list image-view-container item-selector">
 {% for (var i=0; i < data.length; i++) { var item = data[i]; %}
+	{% const item_name = frappe.utils.escape_html(item.name); %}
+	{% const item_title = frappe.utils.escape_html(item.item_name || item.name); %}
 	{% if (i % 4 === 0) { %}<div class="image-view-row">{% } %}
-	<div class="image-view-item" data-name="{{ item.name }}">
+	<div class="image-view-item" data-name="{{ item_name }}">
 		<div class="image-view-header doclist-row">
 			<div class="list-value">
-				<a class="grey list-id" data-name="{{item.name}}"
-					title="{{ item.item_name || item.name}}">
-					{{item.item_name || item.name}}</a>
+				<a class="grey list-id" data-name="{{ item_name }}"
+					title="{{ item_title }}">
+					{{ item_title }}</a>
 			</div>
 		</div>
 		<div class="image-view-body">
-			<a  data-item-code="{{ item.name }}"
-				title="{{ item.item_name || item.name }}"
+			<a  data-item-code="{{ item_name }}"
+				title="{{ item_title }}"
 			>
 				<div class="image-field"
 					style="
@@ -22,11 +24,11 @@
 				>
 					{% if (!item.image) { %}
 					<span class="placeholder-text">
-						{%= frappe.get_abbr(item.item_name || item.name) %}
+						{%= frappe.get_abbr(item_title) %}
 					</span>
 					{% } %}
 					{% if (item.image) { %}
-					<img src="{{ item.image }}" alt="{{item.item_name || item.name}}">
+					<img src="{{ frappe.utils.escape_html(item.image) }}" alt="{{ item_title }}">
 					{% } %}
 				</div>
 			</a>

erpnext/public/js/templates/visual_plant_floor_template.html

@@ -1,5 +1,6 @@
 {% $.each(workstations, (idx, row) => { %}
-	<div class="workstation-wrapper" data-workstation="{{row.name}}">
+	{% const row_workstation_name = frappe.utils.escape_html(row.name); %}
+	<div class="workstation-wrapper" data-workstation="{{row_workstation_name}}">
 		<div class="workstation-status text-left" style="">
 			<span class="indicator-pill no-indicator-dot whitespace-nowrap {{row.color}}" style="margin: 8px 0px 0px 8px;">
 				<span class="workstation-status-title" style="font-size:10px">{{row.status}}</span>
@@ -10,12 +11,12 @@
 				{% if(row.status_image) { %}
 					<img class="workstation-image-cls" src="{{row.status_image}}">
 				{% } else { %}
-					<div class="workstation-image-cls workstation-abbr" style="margin:6px; height:82px">{{frappe.get_abbr(row.name, 2)}}</div>
+					<div class="workstation-image-cls workstation-abbr" style="margin:6px; height:82px">{{frappe.get_abbr(row_workstation_name, 2)}}</div>
 				{% } %}
 			</div>
-			<span class="ellipsis" title="{{row.name}}">
+			<span class="ellipsis" title="{{row_workstation_name}}">
 				<div style="font-size:11px; text-align:center;padding-bottom:8px">{{row.workstation_name}}</div>
 			</span>
 		</div>
 	</div>
-{% }); %}
+{% }); %}

erpnext/selling/report/pending_so_items_for_purchase_request/pending_so_items_for_purchase_request.py

@@ -49,6 +49,29 @@ def get_columns():
 	return columns
 
 
+def apply_representative_lines(rows, sales_orders):
+	"""Fill item_name/description from one real Sales Order Item line per group.
+
+	Both are editable per line, so an order listing the same item twice holds several values per
+	group. Aggregating them sorts text, and MariaDB folds case while PostgreSQL orders by byte
+	value, so the engines pick differently. Take the first line by idx.
+	"""
+	representative = {}
+	if sales_orders:
+		for line in frappe.get_all(
+			"Sales Order Item",
+			filters={"parent": ("in", sales_orders), "docstatus": 1},
+			fields=["parent", "item_code", "item_name", "description"],
+			order_by="idx",
+		):
+			representative.setdefault((line.parent, line.item_code), line)
+
+	for row in rows:
+		line = representative.get((row.name, row.item_code))
+		row.item_name = line.item_name if line else None
+		row.description = line.description if line else None
+
+
 def get_data():
 	so = frappe.qb.DocType("Sales Order")
 	so_item = frappe.qb.DocType("Sales Order Item")
@@ -58,10 +81,9 @@ def get_data():
 		.on(so.name == so_item.parent)
 		.select(
 			so_item.item_code,
-			# non-grouped columns are constant per grouped so.name / item_code -> Max() keeps the
-			# GROUP BY valid on postgres while returning the same value MySQL picked.
-			Max(so_item.item_name).as_("item_name"),
-			Max(so_item.description).as_("description"),
+			# the Sales Order columns are functionally dependent on the grouped so.name, so Max()
+			# returns their single value. item_name/description belong to the line and are editable
+			# per line, so they come from a representative line below.
 			so.name,
 			Max(so.transaction_date).as_("transaction_date"),
 			Max(so.customer).as_("customer"),
@@ -75,6 +97,7 @@ def get_data():
 	)
 
 	sales_orders = [row.name for row in sales_order_entry]
+	apply_representative_lines(sales_order_entry, sales_orders)
 	mr_records = frappe.get_all(
 		"Material Request Item",
 		{"sales_order": ("in", sales_orders), "docstatus": 1},

erpnext/selling/report/quotation_trends/test_quotation_trends.py

@@ -88,6 +88,37 @@ class TestQuotationTrends(ERPNextTestSuite):
 		labels, after = self.run_report(based_on="Customer")
 		self.assertEqual(self._cell(after, "Party", "_Test Customer", amt_col, labels) - before_amt, 300)
 
+	def test_lead_quotation_label_resolves_through_quotation_to(self):
+		"""party_name is a dynamic link, so the label must be resolved via quotation_to.
+
+		Looking the party up in Customer alone leaves a Lead's row blank, and looking in Customer
+		first returns the wrong record when a Lead and a Customer share a name.
+		"""
+		lead_name = "_Test Trends Lead Party"
+		if not frappe.db.exists("Lead", {"lead_name": lead_name}):
+			frappe.get_doc({"doctype": "Lead", "lead_name": lead_name}).insert()
+		lead = frappe.db.get_value("Lead", {"lead_name": lead_name}, ["name", "company_name"], as_dict=True)
+
+		quotation = frappe.new_doc("Quotation")
+		quotation.company = "_Test Company"
+		quotation.transaction_date = TXN_DATE
+		quotation.currency = "INR"
+		quotation.quotation_to = "Lead"
+		quotation.party_name = lead.name
+		quotation.append(
+			"items",
+			{"item_code": "_Test Item", "qty": 1, "rate": 100, "warehouse": "_Test Warehouse - _TC"},
+		)
+		quotation.insert()
+		quotation.submit()
+
+		labels, rows = self.run_report(based_on="Customer")
+		party_idx, name_idx = labels.index("Party"), labels.index("Party Name")
+		lead_rows = [row for row in rows if row[party_idx] == lead.name]
+
+		self.assertEqual(len(lead_rows), 1)
+		self.assertEqual(lead_rows[0][name_idx], lead.company_name or lead_name)
+
 	def test_group_by_chart_matches_table_total_with_mixed_group_sizes(self):
 		# _Test Item is quoted to two customers -> two detail rows under one header row.
 		# _Test Item 2 is quoted to only one customer -> exactly one detail row under its

erpnext/selling/report/sales_order_trends/test_sales_order_trends.py

@@ -31,11 +31,41 @@ class TestSalesOrderTrends(ERPNextTestSuite):
 		self.assertTrue(columns)
 		self.assertTrue(any("_Test Item" in [str(cell) for cell in row] for row in data))
 
+	def test_customer_labels_come_from_the_master_not_a_stored_snapshot(self):
+		"""territory and customer_name must be the Customer master's, not one order's snapshot.
+
+		Both are stored per transaction and editable, so historical orders can hold different values
+		for one customer. Aggregating them with Max() is a text sort, and MariaDB (case-folding) and
+		PostgreSQL (byte order) resolve it differently, so the two engines could label the same row
+		differently. The master's values are functionally dependent on the grouped customer, so they
+		are the same on both engines by construction.
+		"""
+		from erpnext.selling.doctype.sales_order.test_sales_order import make_sales_order
+		from erpnext.selling.report.sales_order_trends.sales_order_trends import execute
+
+		make_sales_order(customer="_Test Customer", item_code="_Test Item", qty=3, rate=100)
+		so2 = make_sales_order(customer="_Test Customer", item_code="_Test Item", qty=2, rate=100)
+		frappe.db.set_value("Sales Order", so2.name, "territory", "_Test Territory Rest Of The World")
+
+		master_territory, master_name = frappe.db.get_value(
+			"Customer", "_Test Customer", ["territory", "customer_name"]
+		)
+
+		columns, data, _chart_none, _chart = execute(
+			{"company": "_Test Company", "period": "Monthly", "based_on": "Customer"}
+		)
+
+		self.assertTrue(columns)
+		customer_rows = [row for row in data if row[0] == "_Test Customer"]
+		self.assertEqual(len(customer_rows), 1)
+		self.assertEqual(customer_rows[0][1], master_name)
+		self.assertEqual(customer_rows[0][2], master_territory)
+
 	def test_customer_with_divergent_stored_territory_stays_one_row(self):
 		# territory (and customer_name) are stored per-transaction fields; historical sales docs can hold a
-		# different value for the same customer. trends groups by t1.customer only and aggregates these with
-		# Max(), so the report stays one row per customer on both MariaDB and Postgres. Grouping by territory
-		# (the pre-fix behaviour) would split the customer into two rows.
+		# different value for the same customer. The report reads both from the Customer master, so it stays
+		# one row per customer on both MariaDB and Postgres. Grouping by the stored territory would split
+		# the customer into two rows.
 		from erpnext.selling.doctype.sales_order.test_sales_order import make_sales_order
 		from erpnext.selling.report.sales_order_trends.sales_order_trends import execute
 

erpnext/stock/deprecated_serial_batch.py

@@ -145,6 +145,9 @@ class DeprecatedBatchNoValuation:
 		if self.sle.name:
 			conditions &= sle.name != self.sle.name
 
+		if getattr(self, "stock_closing_from_datetime", None):
+			conditions &= sle.posting_datetime >= self.stock_closing_from_datetime
+
 		# MariaDB carries a row lock on the grouped query below; on postgres the caller
 		# (calculate_avg_rate) serializes via a txn-scoped advisory lock on (item, warehouse).
 		query = (

erpnext/stock/doctype/item/item.py

@@ -860,7 +860,17 @@ class Item(Document):
 				frappe.throw(_("Item {0} is not a template item.").format(frappe.bold(self.variant_of)))
 
 			if based_on == "Item Attribute":
+				previous_doc = self.get_doc_before_save()
+				saved_attributes = (
+					{(row.attribute, row.attribute_value) for row in previous_doc.attributes}
+					if previous_doc
+					else set()
+				)
+
 				for d in self.attributes:
+					if (d.attribute, d.attribute_value) in saved_attributes:
+						continue
+
 					if not frappe.db.exists(
 						"Item Variant Attribute", {"attribute": d.attribute, "parent": self.variant_of}
 					):

erpnext/stock/doctype/item/test_item.py

@@ -423,6 +423,45 @@ class TestItem(ERPNextTestSuite):
 
 		self.assertRaises(InvalidItemAttributeValueError, attribute.save)
 
+	def test_disabled_attribute_blocks_only_attribute_changes(self):
+		frappe.delete_doc_if_exists("Item", "_Test Disabled Attribute Template-L", force=1)
+		frappe.delete_doc_if_exists("Item", "_Test Disabled Attribute Template", force=1)
+		frappe.delete_doc_if_exists("Item Attribute", "_Test Disabled Size", force=1)
+
+		attribute = frappe.get_doc(
+			{
+				"doctype": "Item Attribute",
+				"attribute_name": "_Test Disabled Size",
+				"item_attribute_values": [
+					{"attribute_value": "Large", "abbr": "L"},
+					{"attribute_value": "Small", "abbr": "S"},
+				],
+			}
+		).insert()
+
+		template = make_item(
+			"_Test Disabled Attribute Template",
+			{
+				"has_variants": 1,
+				"variant_based_on": "Item Attribute",
+				"attributes": [{"attribute": attribute.name}],
+			},
+		)
+
+		variant = create_variant(template.name, {attribute.name: "Large"})
+		variant.save()
+
+		attribute.disabled = 1
+		attribute.save()
+
+		variant.reload()
+		variant.description = "Edited after the attribute was disabled"
+		variant.save()
+
+		variant.reload()
+		variant.attributes[0].attribute_value = "Small"
+		self.assertRaises(frappe.ValidationError, variant.save)
+
 	def test_rename_attribute_value_updates_variants(self):
 		frappe.delete_doc_if_exists("Item", "_Test Variant Item-L", force=1)
 

erpnext/stock/doctype/quality_inspection/quality_inspection.py

@@ -2,13 +2,15 @@
 # License: GNU General Public License v3. See license.txt
 
 
+from math import isfinite
 from typing import Any
 
 import frappe
 from frappe import _
 from frappe.model.document import Document
 from frappe.model.mapper import get_mapped_doc
-from frappe.utils import cint, flt, get_link_to_form, get_number_format_info
+from frappe.utils import cint, flt, get_link_to_form
+from frappe.utils.number_format import NUMBER_FORMAT_MAP, NumberFormat
 
 from erpnext.stock.doctype.quality_inspection_template.quality_inspection_template import (
 	get_template_details,
@@ -84,6 +86,7 @@ class QualityInspection(Document):
 						reading.status = "Accepted"
 
 		if self.readings:
+			self.validate_reading_number_format()
 			self.inspect_and_set_status()
 
 		self.validate_inspection_required()
@@ -281,6 +284,47 @@ class QualityInspection(Document):
 					)
 					break
 
+	def validate_reading_number_format(self):
+		"""Reject newly entered readings that are not numbers in the user's format.
+
+		They would otherwise be misread rather than refused, silently rejecting an
+		inspection whose readings are in fact within the acceptance range. Readings
+		already stored are left alone, so a document entered by a user in one locale
+		stays saveable and submittable by a user in another."""
+		number_format = get_reading_number_format()
+		decimal_str, comma_str = get_reading_separators(number_format)
+		before_save = self.get_doc_before_save()
+
+		for reading in self.readings:
+			if not cint(reading.numeric) or cint(reading.manual_inspection):
+				continue
+
+			stored = before_save and before_save.get("readings", {"name": reading.name})
+			stored = stored[0] if stored else None
+
+			for i in range(1, 11):
+				field = "reading_" + str(i)
+				value = reading.get(field)
+				if value is None or not value.strip():
+					continue
+
+				if stored and stored.get(field) == value:
+					continue
+
+				if parse_reading(value, decimal_str, comma_str) is None:
+					frappe.throw(
+						_(
+							"Row #{0}: Reading {1} {2} is not a valid number in the {3} number format. Use {4} as the decimal separator."
+						).format(
+							reading.idx,
+							i,
+							frappe.bold(value),
+							frappe.bold(number_format.string),
+							frappe.bold(decimal_str),
+						),
+						title=_("Invalid Reading"),
+					)
+
 	def set_status_based_on_acceptance_values(self, reading):
 		if not cint(reading.numeric):
 			reading_value = reading.get("reading_value") or ""
@@ -511,17 +555,61 @@ def make_quality_inspection(source_name: str, target_doc: str | dict | Document
 	return doc
 
 
+def get_reading_number_format() -> NumberFormat:
+	"""Number format the user enters readings in.
+
+	User defaults fall back to the global default, so this is the same format the
+	user's desk formats numbers with."""
+	number_format = frappe.defaults.get_user_default("number_format")
+	if number_format not in NUMBER_FORMAT_MAP:
+		number_format = "#,###.##"
+
+	return NumberFormat.from_string(number_format)
+
+
+def get_reading_separators(number_format: NumberFormat) -> tuple[str, str]:
+	"""Decimal and thousands separator a reading may be written with.
+
+	A format with no decimal separator still has to accept decimal readings, so it
+	falls back to a dot and gives up any grouping that would collide with it."""
+	decimal_str = number_format.decimal_separator or "."
+	comma_str = number_format.thousands_separator
+
+	return decimal_str, "" if comma_str == decimal_str else comma_str
+
+
+def parse_reading(value: str, decimal_str: str, comma_str: str) -> float | None:
+	"""Reading as a float, or None when it is not a number in that format."""
+	value = value.strip()
+	integer_part = value.partition(decimal_str)[0]
+
+	if comma_str and comma_str in integer_part:
+		groups = integer_part.split(comma_str)
+		lead = groups[0][1:] if groups[0][:1] in ("+", "-") else groups[0]
+		if not 1 <= len(lead) <= 3 or len(groups[-1]) != 3:
+			return None
+
+		if any(len(group) not in (2, 3) for group in groups[1:-1]):
+			return None
+
+		value = value.replace(comma_str, "")
+
+	if decimal_str != ".":
+		value = value.replace(decimal_str, ".")
+
+	try:
+		number = float(value)
+	except ValueError:
+		return None
+
+	return number if isfinite(number) else None
+
+
 def parse_float(num: str) -> float:
 	"""Since reading_# fields are `Data` field they might contain number which
 	is representation in user's prefered number format instead of machine
 	readable format. This function converts them to machine readable format."""
 
-	number_format = frappe.db.get_default("number_format") or "#,###.##"
-	decimal_str, comma_str, _number_format_precision = get_number_format_info(number_format)
+	decimal_str, comma_str = get_reading_separators(get_reading_number_format())
 
-	if decimal_str == "," and comma_str == ".":
-		num = num.replace(",", "#$")
-		num = num.replace(".", ",")
-		num = num.replace("#$", ".")
-
-	return flt(num)
+	return flt(parse_reading(num, decimal_str, comma_str))

erpnext/stock/doctype/quality_inspection/test_quality_inspection.py

@@ -1,8 +1,11 @@
 # Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors and Contributors
 # See license.txt
 
+from contextlib import contextmanager
+
 import frappe
 from frappe.utils import nowdate
+from frappe.utils.number_format import NumberFormat
 
 from erpnext.controllers.stock_controller import (
 	QualityInspectionNotSubmittedError,
@@ -12,10 +15,29 @@ from erpnext.controllers.stock_controller import (
 )
 from erpnext.stock.doctype.delivery_note.test_delivery_note import create_delivery_note
 from erpnext.stock.doctype.item.test_item import create_item
+from erpnext.stock.doctype.quality_inspection.quality_inspection import (
+	get_reading_separators,
+	parse_reading,
+)
 from erpnext.stock.doctype.stock_entry.stock_entry_utils import make_stock_entry
 from erpnext.tests.utils import ERPNextTestSuite
 
 
+@contextmanager
+def user_number_format(number_format):
+	"""Temporarily set the session user's own number format."""
+	user = frappe.session.user
+	previous = frappe.db.get_value("DefaultValue", {"parent": user, "defkey": "number_format"}, "defvalue")
+	frappe.defaults.set_user_default("number_format", number_format)
+	try:
+		yield
+	finally:
+		if previous:
+			frappe.defaults.set_user_default("number_format", previous)
+		else:
+			frappe.defaults.clear_user_default("number_format")
+
+
 class TestQualityInspection(ERPNextTestSuite):
 	def setUp(self):
 		super().setUp()
@@ -108,7 +130,6 @@ class TestQualityInspection(ERPNextTestSuite):
 				"acceptance_formula": "mean < 0.9",
 				"reading_1": "0.5",
 				"reading_2": "0.7",
-				"reading_3": "random text",  # check if random string input causes issues
 			},
 			{
 				"specification": "Calcium Content",  # non-numeric reading
@@ -252,6 +273,208 @@ class TestQualityInspection(ERPNextTestSuite):
 		qa.delete()
 		dn.delete()
 
+	def test_non_numeric_reading(self):
+		dn = create_delivery_note(item_code="_Test Item with QA", do_not_submit=True)
+		create_quality_inspection_parameter("Density")
+
+		readings = [
+			{"specification": "Density", "min_value": 1.15, "max_value": 1.20, "reading_1": "random text"}
+		]
+		qa = create_quality_inspection(
+			reference_type="Delivery Note", reference_name=dn.name, readings=readings, do_not_save=True
+		)
+
+		self.assertRaises(frappe.ValidationError, qa.save)
+
+		dn.delete()
+
+	def test_non_numeric_reading_in_formula_based_criteria(self):
+		dn = create_delivery_note(item_code="_Test Item with QA", do_not_submit=True)
+		create_quality_inspection_parameter("Density")
+
+		readings = [
+			{
+				"specification": "Density",
+				"formula_based_criteria": 1,
+				"acceptance_formula": "mean < 0.9",
+				"reading_1": "0.5",
+				"reading_2": "0.7",
+				"reading_3": "random text",
+			}
+		]
+		qa = create_quality_inspection(
+			reference_type="Delivery Note", reference_name=dn.name, readings=readings, do_not_save=True
+		)
+
+		self.assertRaises(frappe.ValidationError, qa.save)
+
+		dn.delete()
+
+	def test_manual_inspection_reading_is_not_number_checked(self):
+		dn = create_delivery_note(item_code="_Test Item with QA", do_not_submit=True)
+		create_quality_inspection_parameter("Density")
+
+		readings = [
+			{
+				"specification": "Density",
+				"manual_inspection": 1,
+				"status": "Accepted",
+				"min_value": 1.15,
+				"max_value": 1.20,
+				"reading_1": "1.15 g/cm3",
+			}
+		]
+		qa = create_quality_inspection(
+			reference_type="Delivery Note", reference_name=dn.name, readings=readings, do_not_save=True
+		)
+		qa.save()
+
+		self.assertEqual(qa.readings[0].status, "Accepted")
+
+		qa.delete()
+		dn.delete()
+
+	def test_reading_in_comma_decimal_number_format(self):
+		dn = create_delivery_note(item_code="_Test Item with QA", do_not_submit=True)
+		create_quality_inspection_parameter("Density")
+
+		readings = [{"specification": "Density", "min_value": 1.15, "max_value": 1.20, "reading_1": "1,15"}]
+		with user_number_format("#.###,##"):
+			qa = create_quality_inspection(
+				reference_type="Delivery Note", reference_name=dn.name, readings=readings, do_not_save=True
+			)
+			qa.save()
+
+			self.assertEqual(qa.readings[0].status, "Accepted")
+			self.assertEqual(qa.status, "Accepted")
+
+		qa.delete()
+		dn.delete()
+
+	def test_reading_in_space_grouped_number_format(self):
+		dn = create_delivery_note(item_code="_Test Item with QA", do_not_submit=True)
+		create_quality_inspection_parameter("Density")
+
+		readings = [{"specification": "Density", "min_value": 1.15, "max_value": 1.20, "reading_1": "1,15"}]
+		with user_number_format("# ###,##"):
+			qa = create_quality_inspection(
+				reference_type="Delivery Note", reference_name=dn.name, readings=readings, do_not_save=True
+			)
+			qa.save()
+
+			self.assertEqual(qa.readings[0].status, "Accepted")
+			self.assertEqual(qa.status, "Accepted")
+
+		qa.delete()
+		dn.delete()
+
+	def test_reading_in_wrong_decimal_number_format(self):
+		dn = create_delivery_note(item_code="_Test Item with QA", do_not_submit=True)
+		create_quality_inspection_parameter("Density")
+
+		readings = [{"specification": "Density", "min_value": 1.15, "max_value": 1.20, "reading_1": "1.15"}]
+		with user_number_format("#.###,##"):
+			qa = create_quality_inspection(
+				reference_type="Delivery Note", reference_name=dn.name, readings=readings, do_not_save=True
+			)
+
+			self.assertRaises(frappe.ValidationError, qa.save)
+
+		dn.delete()
+
+	def test_reading_with_comma_in_dot_decimal_number_format(self):
+		dn = create_delivery_note(item_code="_Test Item with QA", do_not_submit=True)
+		create_quality_inspection_parameter("Density")
+
+		readings = [{"specification": "Density", "min_value": 1.15, "max_value": 1.20, "reading_1": "1,15"}]
+		with user_number_format("#,###.##"):
+			qa = create_quality_inspection(
+				reference_type="Delivery Note", reference_name=dn.name, readings=readings, do_not_save=True
+			)
+
+			self.assertRaises(frappe.ValidationError, qa.save)
+
+		dn.delete()
+
+	@ERPNextTestSuite.change_settings("System Settings", {"number_format": "#,###.##"})
+	def test_reading_number_format_prefers_the_user_over_the_system(self):
+		dn = create_delivery_note(item_code="_Test Item with QA", do_not_submit=True)
+		create_quality_inspection_parameter("Density")
+
+		readings = [{"specification": "Density", "min_value": 1.15, "max_value": 1.20, "reading_1": "1,15"}]
+		with user_number_format("#.###,##"):
+			qa = create_quality_inspection(
+				reference_type="Delivery Note", reference_name=dn.name, readings=readings, do_not_save=True
+			)
+			qa.save()
+
+			self.assertEqual(qa.readings[0].status, "Accepted")
+
+		qa.delete()
+		dn.delete()
+
+	def test_stored_reading_stays_submittable_in_another_number_format(self):
+		dn = create_delivery_note(item_code="_Test Item with QA", do_not_submit=True)
+		create_quality_inspection_parameter("Density")
+
+		readings = [{"specification": "Density", "min_value": 1.15, "max_value": 1.20, "reading_1": "1,15"}]
+		with user_number_format("#.###,##"):
+			qa = create_quality_inspection(
+				reference_type="Delivery Note", reference_name=dn.name, readings=readings, do_not_save=True
+			)
+			qa.save()
+
+		with user_number_format("#,###.##"):
+			qa.reload()
+			qa.submit()
+
+		self.assertEqual(qa.docstatus, 1)
+
+		qa.cancel()
+		qa.delete()
+		dn.delete()
+
+	def test_parse_reading_in_every_number_format(self):
+		accepted = [
+			("#,###.##", "1.15", 1.15),
+			("#,###.##", "1,234.56", 1234.56),
+			("#,##,###.##", "12,34,567.89", 1234567.89),
+			("#,###.###", "1,234.567", 1234.567),
+			("#.###,##", "1,15", 1.15),
+			("#.###,##", "1.234,56", 1234.56),
+			("# ###,##", "1,15", 1.15),
+			("# ###,##", "1.15", 1.15),
+			("# ###,##", "1 234,56", 1234.56),
+			("# ###.##", "1 234.56", 1234.56),
+			("#'###.##", "1'234.56", 1234.56),
+			("#, ###.##", "1, 234.56", 1234.56),
+			("#.########", "1.15", 1.15),
+			("#,###", "1.5", 1.5),
+			("#,###", "1,500", 1500.0),
+			("#.###", "1.5", 1.5),
+			("#.###", "1.500", 1.5),
+			("#,###.##", "-1,234.56", -1234.56),
+		]
+		refused = [
+			("#,###.##", "1,15"),
+			("#.###,##", "1.15"),
+			("#,###.##", "--1.15"),
+			("#,###.##", "1²"),
+			("#,###.##", "nan"),
+			("#,###.##", "random text"),
+			("#,###", "1,50"),
+		]
+
+		for number_format, value, expected in accepted:
+			decimal_str, comma_str = get_reading_separators(NumberFormat.from_string(number_format))
+			with self.subTest(number_format=number_format, value=value):
+				self.assertEqual(parse_reading(value, decimal_str, comma_str), expected)
+
+		for number_format, value in refused:
+			decimal_str, comma_str = get_reading_separators(NumberFormat.from_string(number_format))
+			with self.subTest(number_format=number_format, value=value):
+				self.assertIsNone(parse_reading(value, decimal_str, comma_str))
+
 	def test_delete_quality_inspection_linked_with_stock_entry(self):
 		item_code = create_item("_Test Cicuular Dependecy Item with QA").name
 

erpnext/stock/doctype/serial_and_batch_bundle/test_serial_and_batch_bundle.py

@@ -4,7 +4,7 @@
 import json
 
 import frappe
-from frappe.utils import flt, nowtime, today
+from frappe.utils import add_days, add_to_date, flt, nowtime, today
 
 from erpnext.stock.doctype.item.test_item import make_item
 from erpnext.stock.doctype.serial_and_batch_bundle.serial_and_batch_bundle import (
@@ -1637,3 +1637,190 @@ class TestSerialandBatchBundleLogic(ERPNextTestSuite):
 
 		self.assertNotIn(bundles[1], bundle_wise_serial_nos)
 		self.assertEqual(bundle_wise_serial_nos[bundles[0]], [serial_no])
+
+	@ERPNextTestSuite.change_settings(
+		"Stock Settings", {"auto_create_serial_and_batch_bundle_for_outward": 1}
+	)
+	def test_batchwise_valuation_for_same_posting_datetime_entries(self):
+		# an inward at a different rate and multiple outward rows with the same
+		# item and warehouse share the same posting datetime, the tie-breaking
+		# must include the same-timestamp entries which are already part of the
+		# ledger and must not let the outward rows count each other
+		item_code = make_item(
+			"Test Batchwise Same Posting Datetime Item 1",
+			properties={
+				"is_stock_item": 1,
+				"has_batch_no": 1,
+				"create_new_batch": 1,
+				"batch_number_series": "TBSPD-ITEM1-.#####",
+				"valuation_method": "FIFO",
+			},
+		).name
+
+		warehouse = "_Test Warehouse - _TC"
+
+		receipt = make_stock_entry(
+			item_code=item_code,
+			qty=10,
+			rate=100,
+			target=warehouse,
+			posting_date=add_days(today(), -5),
+			posting_time="12:00:00",
+		)
+
+		batch_no = get_batch_from_bundle(receipt.items[0].serial_and_batch_bundle)
+		self.assertTrue(frappe.db.get_value("Batch", batch_no, "use_batchwise_valuation"))
+
+		# same posting datetime as the outward rows below, at a different rate
+		make_stock_entry(
+			item_code=item_code,
+			qty=20,
+			rate=250,
+			target=warehouse,
+			batch_no=batch_no,
+			use_serial_batch_fields=1,
+			posting_date=add_days(today(), -3),
+			posting_time="12:00:00",
+		)
+
+		issue = make_stock_entry(
+			item_code=item_code,
+			qty=2,
+			source=warehouse,
+			posting_date=add_days(today(), -3),
+			posting_time="12:00:00",
+			do_not_save=True,
+		)
+
+		for qty in [3, 4]:
+			issue.append(
+				"items",
+				{
+					"item_code": item_code,
+					"s_warehouse": warehouse,
+					"qty": qty,
+					"conversion_factor": 1,
+				},
+			)
+
+		issue.save()
+		issue.submit()
+
+		# (10 * 100 + 20 * 250) / 30 = 200
+		self.assert_batchwise_outgoing_rate(item_code, outgoing_rate=200.0, balance_value=4200.0)
+
+		# backdated receipt reposts the same posting datetime cluster
+		make_stock_entry(
+			item_code=item_code,
+			qty=10,
+			rate=100,
+			target=warehouse,
+			batch_no=batch_no,
+			use_serial_batch_fields=1,
+			posting_date=add_days(today(), -4),
+			posting_time="12:00:00",
+		)
+
+		# (20 * 100 + 20 * 250) / 40 = 175
+		self.assert_batchwise_outgoing_rate(item_code, outgoing_rate=175.0, balance_value=5425.0)
+
+	@ERPNextTestSuite.change_settings(
+		"Stock Settings", {"auto_create_serial_and_batch_bundle_for_outward": 1}
+	)
+	def test_batchwise_valuation_when_bundle_created_before_the_sle(self):
+		# a bundle can be created (drafted) much before / after its SLE, the
+		# tie-breaking for the same posting datetime entries must follow the
+		# SLE creation and not the bundle creation
+		item_code = make_item(
+			"Test Batchwise Same Posting Datetime Item 2",
+			properties={
+				"is_stock_item": 1,
+				"has_batch_no": 1,
+				"create_new_batch": 1,
+				"batch_number_series": "TBSPD-ITEM2-.#####",
+				"valuation_method": "FIFO",
+			},
+		).name
+
+		warehouse = "_Test Warehouse - _TC"
+
+		receipt = make_stock_entry(
+			item_code=item_code,
+			qty=10,
+			rate=100,
+			target=warehouse,
+			posting_date=add_days(today(), -5),
+			posting_time="12:00:00",
+		)
+
+		batch_no = get_batch_from_bundle(receipt.items[0].serial_and_batch_bundle)
+
+		# inward at a different rate, same posting datetime as the outward below
+		inward = make_stock_entry(
+			item_code=item_code,
+			qty=10,
+			rate=200,
+			target=warehouse,
+			batch_no=batch_no,
+			use_serial_batch_fields=1,
+			posting_date=add_days(today(), -3),
+			posting_time="12:00:00",
+		)
+
+		outward = make_stock_entry(
+			item_code=item_code,
+			qty=10,
+			source=warehouse,
+			posting_date=add_days(today(), -3),
+			posting_time="12:00:00",
+		)
+
+		# simulate the inward's bundle drafted after the outward's SLE, the
+		# bundle creation timeline no longer matches the SLE creation timeline
+		outward_sle_creation = frappe.db.get_value(
+			"Stock Ledger Entry",
+			{"voucher_no": outward.name, "is_cancelled": 0},
+			"creation",
+		)
+
+		frappe.db.set_value(
+			"Serial and Batch Bundle",
+			inward.items[0].serial_and_batch_bundle,
+			"creation",
+			add_to_date(outward_sle_creation, minutes=30),
+			update_modified=False,
+		)
+
+		repost = frappe.get_doc(
+			{
+				"doctype": "Repost Item Valuation",
+				"based_on": "Item and Warehouse",
+				"item_code": item_code,
+				"warehouse": warehouse,
+				"posting_date": add_days(today(), -6),
+				"posting_time": "00:00:00",
+				"allow_negative_stock": 1,
+			}
+		)
+
+		repost.submit()
+
+		# (10 * 100 + 10 * 200) / 20 = 150, the inward precedes the outward as
+		# per the SLE creation even though its bundle was created afterwards
+		self.assert_batchwise_outgoing_rate(item_code, outgoing_rate=150.0, balance_value=1500.0)
+
+	def assert_batchwise_outgoing_rate(self, item_code, outgoing_rate, balance_value):
+		sl_entries = frappe.get_all(
+			"Stock Ledger Entry",
+			filters={"item_code": item_code, "is_cancelled": 0},
+			fields=["actual_qty", "stock_value_difference", "stock_value"],
+			order_by="posting_datetime, creation",
+		)
+
+		for sle in sl_entries:
+			if sle.actual_qty > 0:
+				continue
+
+			self.assertEqual(flt(sle.stock_value_difference, 2), flt(sle.actual_qty * outgoing_rate, 2))
+
+		self.assertEqual(flt(sl_entries[-1].stock_value, 2), flt(balance_value, 2))

erpnext/stock/doctype/stock_closing_entry/stock_closing_entry.py

@@ -9,9 +9,51 @@ from frappe.desk.form.load import get_attachments
 from frappe.model.document import Document
 from frappe.utils import add_days, get_date_str, get_link_to_form, nowtime, parse_json
 from frappe.utils.background_jobs import enqueue
+from frappe.utils.caching import request_cache
 
 from erpnext.stock.doctype.inventory_dimension.inventory_dimension import get_inventory_dimensions
 
+SCOPE_FIELDS = ("warehouse", "item_code", "item_group", "warehouse_type")
+
+
+def apply_unscoped_filters(filters):
+	meta = frappe.get_meta("Stock Closing Entry")
+	for fieldname in SCOPE_FIELDS:
+		if meta.has_field(fieldname):
+			filters[fieldname] = ("is", "not set")
+
+	return filters
+
+
+def get_closing_entry_for_closed_period(company):
+	closed_upto = frappe.db.get_value(
+		"Period Closing Voucher", {"docstatus": 1, "company": company}, [{"MAX": "period_end_date"}]
+	)
+	if not closed_upto:
+		return None
+
+	return _get_completed_closing_entry(company, str(closed_upto))
+
+
+@request_cache
+def _get_completed_closing_entry(company, closed_upto):
+	filters = apply_unscoped_filters(
+		{
+			"company": company,
+			"docstatus": 1,
+			"status": "Completed",
+			"to_date": ("<=", closed_upto),
+		}
+	)
+
+	return frappe.db.get_value(
+		"Stock Closing Entry",
+		filters,
+		["name", "to_date"],
+		order_by="to_date desc",
+		as_dict=True,
+	)
+
 
 class StockClosingEntry(Document):
 	# begin: auto-generated types
@@ -66,7 +108,7 @@ class StockClosingEntry(Document):
 			)
 		)
 
-		for fieldname in ["warehouse", "item_code", "item_group", "warehouse_type"]:
+		for fieldname in SCOPE_FIELDS:
 			if self.get(fieldname):
 				query = query.where(table[fieldname] == self.get(fieldname))
 
@@ -84,14 +126,30 @@ class StockClosingEntry(Document):
 		self.enqueue_job()
 
 	def on_cancel(self):
+		self.validate_closed_period_lock()
 		self.set_status(save=True)
 		self.remove_stock_closing()
 
+	def validate_closed_period_lock(self):
+		pcv = frappe.db.get_value(
+			"Period Closing Voucher",
+			{"company": self.company, "docstatus": 1, "period_end_date": (">=", self.to_date)},
+			"name",
+		)
+
+		if pcv:
+			frappe.throw(
+				_(
+					"Stock Closing Entry {0} belongs to a closed accounting period. Cancel the Period Closing Voucher {1} first."
+				).format(self.name, get_link_to_form("Period Closing Voucher", pcv)),
+				title=_("Closed Period"),
+			)
+
 	def remove_stock_closing(self):
 		table = frappe.qb.DocType("Stock Closing Balance")
 		frappe.qb.from_(table).delete().where(table.stock_closing_entry == self.name).run()
 
-	@frappe.whitelist()
+	@frappe.whitelist(methods=["POST"])
 	def enqueue_job(self):
 		self.db_set("status", "In Progress")
 		enqueue(prepare_closing_stock_balance, name=self.name, queue="long", timeout=1500)
@@ -101,8 +159,9 @@ class StockClosingEntry(Document):
 			).format(self.name)
 		)
 
-	@frappe.whitelist()
+	@frappe.whitelist(methods=["POST"])
 	def regenerate_closing_balance(self):
+		self.validate_closed_period_lock()
 		self.remove_stock_closing()
 		self.enqueue_job()
 

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

@@ -2,7 +2,7 @@ from collections import defaultdict
 
 import frappe
 from frappe import _
-from frappe.query_builder.functions import Max, Min, NullIf, Sum
+from frappe.query_builder.functions import Min, NullIf, Sum
 from frappe.utils import flt
 
 from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos
@@ -348,35 +348,16 @@ class DisassembleStockEntry(BaseStockEntry):
 				.run(as_dict=True)
 			)
 
-		# Aggregating across all Manufacture entries of the work order, one row per item_code.
-		# The non-grouped columns are constant per item_code in practice (an item plays one role with
-		# one uom/warehouse across the WO's manufacture entries); Max() keeps the GROUP BY valid on
-		# postgres while returning the value MySQL picked arbitrarily, preserving the one-row-per-item
-		# shape the disassembly expects.
-		return (
+		# Aggregate in stock UOM: qty is expressed in each row's selected UOM and cannot be added
+		# when manufacture entries use different UOMs for the same item. basic_rate is also per
+		# stock UOM, so weight it by transfer_qty. Manufacture rows always carry positive stock
+		# qty, so NullIf only guards a theoretical /0.
+		rows = (
 			query.select(
-				Sum(SED.qty).as_("qty"),
-				Sum(SED.transfer_qty).as_("transfer_qty"),
 				SED.item_code,
-				Max(SED.item_name).as_("item_name"),
-				Max(SED.description).as_("description"),
-				Max(SED.stock_uom).as_("stock_uom"),
-				Max(SED.uom).as_("uom"),
-				# qty-weighted average so consolidating an item across manufacture entries at different
-				# valuation rates values the summed qty correctly (Max would bias the rate high).
-				# Manufacture rows always carry positive qty, so NullIf only guards a theoretical /0.
-				(Sum(SED.basic_rate * SED.qty) / NullIf(Sum(SED.qty), 0)).as_("basic_rate"),
-				Max(SED.conversion_factor).as_("conversion_factor"),
-				Max(SED.is_finished_item).as_("is_finished_item"),
-				Max(SED.secondary_item_type).as_("secondary_item_type"),
-				Max(SED.is_legacy_scrap_item).as_("is_legacy_scrap_item"),
-				Max(SED.bom_secondary_item).as_("bom_secondary_item"),
-				Max(SED.batch_no).as_("batch_no"),
-				Max(SED.serial_no).as_("serial_no"),
-				Max(SED.use_serial_batch_fields).as_("use_serial_batch_fields"),
-				Max(SED.s_warehouse).as_("s_warehouse"),
-				Max(SED.t_warehouse).as_("t_warehouse"),
-				Max(SED.bom_no).as_("bom_no"),
+				Sum(SED.transfer_qty).as_("qty"),
+				Sum(SED.transfer_qty).as_("transfer_qty"),
+				(Sum(SED.basic_rate * SED.transfer_qty) / NullIf(Sum(SED.transfer_qty), 0)).as_("basic_rate"),
 			)
 			.where(SE.purpose == "Manufacture")
 			.where(SE.work_order == self.doc.work_order)
@@ -385,6 +366,61 @@ class DisassembleStockEntry(BaseStockEntry):
 			.run(as_dict=True)
 		)
 
+		representative = self.get_representative_manufacture_rows()
+		for row in rows:
+			row.update(representative.get(row.item_code) or {})
+			row.uom = row.stock_uom
+			row.conversion_factor = 1
+
+		return rows
+
+	def get_representative_manufacture_rows(self):
+		"""Earliest posted line per item across the work order's Manufacture entries.
+
+		The disassembly wants one row per item, but some descriptive columns describe a line, not
+		an item: batch_no and serial_no only mean something beside their warehouse, and
+		is_finished_item decides whether the row is the output or an input. Aggregating each column
+		on its own can pair values from different lines into a row that was never posted, so take
+		the columns from a single real line instead. UOM is normalized separately to stock UOM.
+		"""
+		SE = frappe.qb.DocType("Stock Entry")
+		SED = frappe.qb.DocType("Stock Entry Detail")
+
+		lines = (
+			frappe.qb.from_(SED)
+			.join(SE)
+			.on(SED.parent == SE.name)
+			.select(
+				SED.item_code,
+				SED.item_name,
+				SED.description,
+				SED.stock_uom,
+				SED.is_finished_item,
+				SED.secondary_item_type,
+				SED.is_legacy_scrap_item,
+				SED.bom_secondary_item,
+				SED.batch_no,
+				SED.serial_no,
+				SED.use_serial_batch_fields,
+				SED.s_warehouse,
+				SED.t_warehouse,
+				SED.bom_no,
+			)
+			.where(
+				(SE.docstatus == 1) & (SE.purpose == "Manufacture") & (SE.work_order == self.doc.work_order)
+			)
+			.orderby(SE.creation)
+			.orderby(SE.name)
+			.orderby(SED.idx)
+			.run(as_dict=True)
+		)
+
+		representative = {}
+		for line in lines:
+			representative.setdefault(line.item_code, line)
+
+		return representative
+
 	def on_submit(self):
 		self.set_serial_batch_for_disassembly()
 		self.update_disassembled_order()

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

@@ -1007,11 +1007,9 @@ def get_secondary_items_from_job_card(work_order, jc_name=None):
 		.select(
 			Sum(job_card_secondary_item.stock_qty).as_("stock_qty"),
 			job_card_secondary_item.item_code,
-			# non-grouped columns are item attributes / the secondary-item BOM link, constant per
-			# grouped (item_code, secondary_item_type) -> Max() keeps the GROUP BY valid on postgres
-			# while returning the value MySQL picked arbitrarily.
-			Max(job_card_secondary_item.item_name).as_("item_name"),
-			Max(job_card_secondary_item.description).as_("description"),
+			# stock_uom and the secondary-item BOM link are constant per grouped
+			# (item_code, secondary_item_type) -> Max() returns their single value. item_name and
+			# description are editable per line, so they come from a representative line below.
 			Max(job_card_secondary_item.stock_uom).as_("stock_uom"),
 			job_card_secondary_item.secondary_item_type,
 			Max(job_card_secondary_item.bom_secondary_item).as_("bom_secondary_item"),
@@ -1030,7 +1028,41 @@ def get_secondary_items_from_job_card(work_order, jc_name=None):
 	if jc_name:
 		secondary_items = secondary_items.where(job_card.name == jc_name)
 
-	return secondary_items.run(as_dict=1)
+	rows = secondary_items.run(as_dict=1)
+	apply_representative_secondary_lines(rows, work_order, jc_name)
+	return rows
+
+
+def apply_representative_secondary_lines(rows, work_order, jc_name=None):
+	"""Fill item_name/description from one real Job Card Secondary Item line per group.
+
+	Both are editable per line, so the same secondary item across a work order's job cards can
+	carry several values per group. Aggregating them sorts text, and MariaDB folds case while
+	PostgreSQL orders by byte value, so the engines pick differently.
+	"""
+	job_cards = frappe.get_all(
+		"Job Card",
+		filters={"work_order": work_order, "docstatus": 1, **({"name": jc_name} if jc_name else {})},
+		pluck="name",
+	)
+
+	representative = {}
+	if job_cards:
+		for line in frappe.get_all(
+			"Job Card Secondary Item",
+			filters={"parent": ("in", job_cards)},
+			# idx first, so the rule really is "first by idx"; creation breaks ties across job cards.
+			# Never order by parent -- the Job Card name is text, and sorting text is the divergence
+			# this is here to avoid.
+			fields=["item_code", "secondary_item_type", "item_name", "description"],
+			order_by="idx, creation",
+		):
+			representative.setdefault((line.item_code, line.secondary_item_type), line)
+
+	for row in rows:
+		line = representative.get((row.item_code, row.secondary_item_type))
+		row.item_name = line.item_name if line else None
+		row.description = line.description if line else None
 
 
 def get_previous_operation_output_sn_batch(work_order, item_code, warehouse):

erpnext/stock/doctype/stock_entry/stock_entry.py

@@ -70,6 +70,15 @@ from erpnext.controllers.subcontracting_inward_controller import SubcontractingI
 form_grid_templates = {"items": "templates/form_grid/stock_entry_grid.html"}
 
 
+def is_costed_out_of_finished_item(row) -> bool:
+	"""Whether the row takes its value out of the finished good instead of adding to it.
+
+	A secondary item that is not linked to a BOM has no cost allocation of its own, so it is
+	valued the way the legacy scrap item was: its cost is deducted from the finished good.
+	"""
+	return bool(row.is_legacy_scrap_item or (row.secondary_item_type and not row.bom_secondary_item))
+
+
 class StockEntry(StockController, SubcontractingInwardController):
 	# begin: auto-generated types
 	# This code is auto-generated. Do not modify anything in this block.
@@ -563,8 +572,11 @@ class StockEntry(StockController, SubcontractingInwardController):
 			frappe.get_cached_value("BOM", self.bom_no, "cost_allocation_per") if self.bom_no else None
 		)
 
+		secondary_items_cost_basis = self.get_secondary_items_cost_basis(outgoing_items_cost)
+
 		zero_valuation_items = []
-		for d in self.get("items"):
+		finished_items_last = sorted(self.get("items"), key=lambda row: cint(row.is_finished_item))
+		for d in finished_items_last:
 			if d.s_warehouse or d.set_basic_rate_manually:
 				continue
 
@@ -581,11 +593,26 @@ class StockEntry(StockController, SubcontractingInwardController):
 				zero_valuation_items,
 				bom_cost_allocation_per,
 				has_consumption_basis,
+				secondary_items_cost_basis,
 			)
 
 		if zero_valuation_items:
 			self._notify_zero_valuation_rate(zero_valuation_items)
 
+	def get_secondary_items_cost_basis(self, outgoing_items_cost) -> float:
+		"""The cost a BOM allocation splits: the consumed rows, or the entry that replaced them."""
+		if outgoing_items_cost or self.purpose != "Manufacture" or not self.work_order:
+			return outgoing_items_cost
+
+		settings = frappe.get_single("Manufacturing Settings")
+		if not (settings.material_consumption and settings.get_rm_cost_from_consumption_entry):
+			return outgoing_items_cost
+
+		if not self.get_consumption_entries():
+			return outgoing_items_cost
+
+		return self._fetch_consumption_entry_cost()
+
 	def has_consumption_basis(self) -> bool:
 		"""Whether the cost of the consumed items is known, even when that cost is zero."""
 		if any(d.s_warehouse for d in self.get("items")):
@@ -619,8 +646,9 @@ class StockEntry(StockController, SubcontractingInwardController):
 		zero_valuation_items,
 		bom_cost_allocation_per=None,
 		has_consumption_basis=False,
+		secondary_items_cost_basis=0,
 	):
-		rate_derived_from_consumption = False
+		has_derived_rate = False
 
 		if d.allow_zero_valuation_rate and d.basic_rate and self.purpose != "Receive from Customer":
 			d.basic_rate = 0.0
@@ -630,26 +658,25 @@ class StockEntry(StockController, SubcontractingInwardController):
 				d.basic_rate = self.get_basic_rate_for_manufactured_item(
 					d.transfer_qty, outgoing_items_cost, has_consumption_basis
 				)
-				rate_derived_from_consumption = has_consumption_basis
+				has_derived_rate = has_consumption_basis
 			elif self.purpose == "Repack":
 				d.basic_rate = self.get_basic_rate_for_repacked_items(d.transfer_qty, outgoing_items_cost)
 				# Repack rate comes from consumed source-warehouse rows, not consumption entries
-				rate_derived_from_consumption = any(item.s_warehouse for item in self.get("items"))
+				has_derived_rate = any(item.s_warehouse for item in self.get("items"))
 
 			if self.bom_no:
 				d.basic_rate *= bom_cost_allocation_per / 100
 		elif d.secondary_item_type and d.bom_secondary_item:
-			cost_allocation_per = frappe.get_value(
-				"BOM Secondary Item", d.bom_secondary_item, "cost_allocation_per"
+			cost_allocation_per = flt(
+				frappe.get_value("BOM Secondary Item", d.bom_secondary_item, "cost_allocation_per")
 			)
-			# Only recalculate when cost is actually allocated; otherwise preserve the
-			# user-entered rate (or fall through to get_valuation_rate below)
-			if cost_allocation_per and flt(d.transfer_qty):
-				d.basic_rate = (outgoing_items_cost * (cost_allocation_per / 100)) / d.transfer_qty
+			if flt(d.transfer_qty):
+				d.basic_rate = (secondary_items_cost_basis * (cost_allocation_per / 100)) / d.transfer_qty
+				has_derived_rate = True
 
-		# A rate of zero derived from the consumed items is their actual cost, not a missing
-		# rate. Falling back to the item's valuation here would value free inputs as output.
-		if not d.basic_rate and not d.allow_zero_valuation_rate and not rate_derived_from_consumption:
+		# A rate of zero that was derived rather than left unset is a real cost. Falling back to
+		# the item's valuation here would value free inputs, or an unallocated row, as output.
+		if not d.basic_rate and not d.allow_zero_valuation_rate and not has_derived_rate:
 			d.basic_rate = get_valuation_rate(
 				d.item_code,
 				d.t_warehouse,
@@ -736,7 +763,9 @@ class StockEntry(StockController, SubcontractingInwardController):
 		self, finished_item_qty, outgoing_items_cost=0, has_consumption_basis=False
 	) -> float:
 		settings = frappe.get_single("Manufacturing Settings")
-		scrap_items_cost = sum([flt(d.basic_amount) for d in self.get("items") if d.is_legacy_scrap_item])
+		scrap_items_cost = sum(
+			[flt(d.basic_amount) for d in self.get("items") if is_costed_out_of_finished_item(d)]
+		)
 
 		if settings.material_consumption:
 			outgoing_items_cost = self._get_rm_cost_for_manufacture(
@@ -901,7 +930,9 @@ class StockEntry(StockController, SubcontractingInwardController):
 
 		for d in self.items:
 			if d.t_warehouse and not d.s_warehouse:
-				if self.purpose == "Repack" or d.item_code == finished_item:
+				if d.secondary_item_type or d.is_legacy_scrap_item:
+					d.is_finished_item = 0
+				elif self.purpose == "Repack" or d.item_code == finished_item:
 					d.is_finished_item = 1
 			else:
 				d.is_finished_item = 0

erpnext/stock/doctype/stock_entry/test_stock_entry.py

@@ -7,6 +7,7 @@ from frappe.utils import add_days, cstr, flt, get_time, getdate, nowtime, today
 
 from erpnext.accounts.doctype.account.test_account import get_inventory_account
 from erpnext.controllers.accounts_controller import InvalidQtyError
+from erpnext.exceptions import QualityInspectionRequiredError
 from erpnext.stock.doctype.item.test_item import (
 	create_item,
 	make_item,
@@ -2728,6 +2729,254 @@ class TestStockEntry(ERPNextTestSuite):
 		self.assertEqual(fg_sle.incoming_rate, 0)
 		self.assertEqual(fg_sle.stock_value_difference, 0)
 
+	def test_secondary_item_type_does_not_waive_inspection_outside_manufacturing(self):
+		"""A stray secondary item type must not let a QI-required item through a receipt."""
+		item = make_item(
+			properties={
+				"is_stock_item": 1,
+				"valuation_rate": 50,
+				"inspection_required_before_purchase": 1,
+			}
+		).name
+
+		def receipt(secondary_item_type):
+			se = frappe.new_doc("Stock Entry")
+			se.purpose = se.stock_entry_type = "Material Receipt"
+			se.company = "_Test Company"
+			se.inspection_required = 1
+			se.append(
+				"items",
+				{
+					"item_code": item,
+					"t_warehouse": "_Test Warehouse - _TC",
+					"qty": 10,
+					"conversion_factor": 1,
+					"secondary_item_type": secondary_item_type,
+				},
+			)
+			return se
+
+		self.assertRaises(QualityInspectionRequiredError, receipt("").submit)
+		self.assertRaises(QualityInspectionRequiredError, receipt("Scrap").submit)
+
+	def test_manufacture_balances_secondary_item_added_without_a_bom(self):
+		"""A secondary item with no BOM link is costed out of the finished good, as legacy scrap was."""
+		rm_item = make_item(properties={"is_stock_item": 1}).name
+		fg_item = make_item(properties={"is_stock_item": 1}).name
+		scrap_item = make_item(properties={"is_stock_item": 1, "valuation_rate": 20}).name
+		warehouse = "_Test Warehouse - _TC"
+
+		make_stock_entry(item_code=rm_item, target=warehouse, qty=10, basic_rate=100)
+
+		se = frappe.new_doc("Stock Entry")
+		se.purpose = se.stock_entry_type = "Manufacture"
+		se.company = "_Test Company"
+		se.append(
+			"items", {"item_code": rm_item, "s_warehouse": warehouse, "qty": 10, "conversion_factor": 1}
+		)
+		se.append(
+			"items",
+			{
+				"item_code": fg_item,
+				"t_warehouse": warehouse,
+				"qty": 10,
+				"is_finished_item": 1,
+				"conversion_factor": 1,
+			},
+		)
+		se.append(
+			"items",
+			{
+				"item_code": scrap_item,
+				"t_warehouse": warehouse,
+				"qty": 5,
+				"secondary_item_type": "Scrap",
+				"conversion_factor": 1,
+			},
+		)
+		se.save()
+
+		scrap_row = se.items[2]
+		self.assertEqual(flt(scrap_row.basic_rate), 20.0)
+		self.assertEqual(flt(scrap_row.basic_amount), 100.0)
+
+		fg_row = se.items[1]
+		self.assertEqual(flt(fg_row.basic_rate), 90.0)
+		self.assertEqual(flt(fg_row.basic_amount), 900.0)
+
+		self.assertEqual(flt(se.total_incoming_value), 1000.0)
+		self.assertEqual(flt(se.total_outgoing_value), 1000.0)
+		self.assertEqual(flt(se.value_difference), 0.0)
+
+	def test_repack_allocates_cost_to_secondary_item(self):
+		"""A Repack secondary item takes its own BOM share, not the finished good's."""
+		rm_item = make_item(properties={"is_stock_item": 1, "valuation_rate": 100}).name
+		fg_item = make_item(properties={"is_stock_item": 1}).name
+		scrap_item = make_item(properties={"is_stock_item": 1, "valuation_rate": 20}).name
+		warehouse = "_Test Warehouse - _TC"
+
+		bom = frappe.get_doc(
+			{
+				"doctype": "BOM",
+				"item": fg_item,
+				"currency": "INR",
+				"quantity": 10,
+				"company": "_Test Company",
+			}
+		)
+		bom.append("items", {"item_code": rm_item, "qty": 10})
+		bom.append(
+			"secondary_items",
+			{
+				"secondary_item_type": "Scrap",
+				"item_code": scrap_item,
+				"item_name": scrap_item,
+				"qty": 5,
+				"cost_allocation_per": 25,
+				"process_loss_per": 0,
+			},
+		)
+		bom.insert()
+		bom.submit()
+		self.assertEqual(flt(bom.cost_allocation_per), 75.0)
+
+		make_stock_entry(item_code=rm_item, target=warehouse, qty=100, basic_rate=100)
+
+		se = frappe.new_doc("Stock Entry")
+		se.purpose = se.stock_entry_type = "Repack"
+		se.company = "_Test Company"
+		se.from_bom = 1
+		se.bom_no = bom.name
+		se.fg_completed_qty = 10
+		se.from_warehouse = warehouse
+		se.to_warehouse = warehouse
+		se.get_items()
+		se.save()
+
+		fg_row = next(d for d in se.items if d.is_finished_item)
+		scrap_row = next(d for d in se.items if d.secondary_item_type)
+
+		self.assertFalse(scrap_row.is_finished_item)
+		self.assertEqual(flt(scrap_row.basic_amount), 250.0)
+		self.assertEqual(flt(fg_row.basic_amount), 750.0)
+
+		self.assertEqual(flt(se.total_incoming_value), 1000.0)
+		self.assertEqual(flt(se.total_outgoing_value), 1000.0)
+		self.assertEqual(flt(se.value_difference), 0.0)
+
+	def test_secondary_item_with_zero_cost_allocation_carries_no_value(self):
+		"""A BOM that allocates 0% to a secondary item gives the finished good everything."""
+		from erpnext.manufacturing.doctype.work_order.test_work_order import make_wo_order_test_record
+		from erpnext.manufacturing.doctype.work_order.work_order import (
+			make_stock_entry as make_stock_entry_from_wo,
+		)
+
+		rm_item = make_item(properties={"is_stock_item": 1, "valuation_rate": 100}).name
+		fg_item = make_item(properties={"is_stock_item": 1}).name
+		scrap_item = make_item(properties={"is_stock_item": 1, "valuation_rate": 20}).name
+		warehouse = "_Test Warehouse - _TC"
+
+		bom = frappe.get_doc(
+			{
+				"doctype": "BOM",
+				"item": fg_item,
+				"currency": "INR",
+				"quantity": 10,
+				"company": "_Test Company",
+			}
+		)
+		bom.append("items", {"item_code": rm_item, "qty": 10})
+		bom.append(
+			"secondary_items",
+			{
+				"secondary_item_type": "Scrap",
+				"item_code": scrap_item,
+				"item_name": scrap_item,
+				"qty": 5,
+				"cost_allocation_per": 0,
+				"process_loss_per": 0,
+			},
+		)
+		bom.insert()
+		bom.submit()
+		self.assertEqual(flt(bom.cost_allocation_per), 100.0)
+
+		make_stock_entry(item_code=rm_item, target=warehouse, qty=100, basic_rate=100)
+		wo = make_wo_order_test_record(
+			production_item=fg_item, bom_no=bom.name, qty=10, skip_transfer=1, source_warehouse=warehouse
+		)
+
+		se = frappe.get_doc(make_stock_entry_from_wo(wo.name, "Manufacture", 10))
+		se.save()
+
+		scrap_row = next(d for d in se.items if d.secondary_item_type)
+		fg_row = next(d for d in se.items if d.is_finished_item)
+
+		self.assertEqual(flt(scrap_row.basic_rate), 0.0)
+		self.assertEqual(flt(scrap_row.basic_amount), 0.0)
+		self.assertEqual(flt(fg_row.basic_amount), 1000.0)
+		self.assertEqual(flt(se.value_difference), 0.0)
+
+	@ERPNextTestSuite.change_settings(
+		"Manufacturing Settings", {"material_consumption": 1, "get_rm_cost_from_consumption_entry": 1}
+	)
+	def test_secondary_item_allocation_uses_consumption_entry_cost(self):
+		"""A BOM allocation splits the consumption entry's cost, not an empty set of consumed rows."""
+		from erpnext.manufacturing.doctype.work_order.test_work_order import make_wo_order_test_record
+		from erpnext.manufacturing.doctype.work_order.work_order import (
+			make_stock_entry as make_stock_entry_from_wo,
+		)
+
+		rm_item = make_item(properties={"is_stock_item": 1, "valuation_rate": 100}).name
+		fg_item = make_item(properties={"is_stock_item": 1}).name
+		scrap_item = make_item(properties={"is_stock_item": 1, "valuation_rate": 20}).name
+		warehouse = "_Test Warehouse - _TC"
+
+		bom = frappe.get_doc(
+			{
+				"doctype": "BOM",
+				"item": fg_item,
+				"currency": "INR",
+				"quantity": 10,
+				"company": "_Test Company",
+			}
+		)
+		bom.append("items", {"item_code": rm_item, "qty": 10})
+		bom.append(
+			"secondary_items",
+			{
+				"secondary_item_type": "Scrap",
+				"item_code": scrap_item,
+				"item_name": scrap_item,
+				"qty": 5,
+				"cost_allocation_per": 25,
+				"process_loss_per": 0,
+			},
+		)
+		bom.insert()
+		bom.submit()
+
+		make_stock_entry(item_code=rm_item, target=warehouse, qty=100, basic_rate=100)
+		wo = make_wo_order_test_record(
+			production_item=fg_item, bom_no=bom.name, qty=10, skip_transfer=1, source_warehouse=warehouse
+		)
+
+		consumption = frappe.get_doc(
+			make_stock_entry_from_wo(wo.name, "Material Consumption for Manufacture", 10)
+		)
+		consumption.submit()
+		self.assertEqual(flt(consumption.total_outgoing_value), 1000.0)
+
+		se = frappe.get_doc(make_stock_entry_from_wo(wo.name, "Manufacture", 10))
+		se.save()
+
+		scrap_row = next(d for d in se.items if d.secondary_item_type)
+		fg_row = next(d for d in se.items if d.is_finished_item)
+
+		self.assertEqual(flt(fg_row.basic_amount), 750.0)
+		self.assertEqual(flt(scrap_row.basic_amount), 250.0)
+		self.assertEqual(flt(se.total_incoming_value), 1000.0)
+
 	def _make_wo_for_free_raw_material(self, rm_item, fg_item, bom_no):
 		from erpnext.manufacturing.doctype.work_order.test_work_order import make_wo_order_test_record
 		from erpnext.manufacturing.doctype.work_order.work_order import (

erpnext/stock/doctype/stock_settings/stock_settings.json

@@ -125,7 +125,8 @@
    "description": "The percentage you are allowed to receive or deliver more against the quantity ordered. For example, if you have ordered 100 units, and your Allowance is 10%, then you are allowed to receive 110 units.",
    "fieldname": "over_delivery_receipt_allowance",
    "fieldtype": "Float",
-   "label": "Over Delivery/Receipt Allowance (%)"
+   "label": "Over Delivery/Receipt Allowance (%)",
+   "non_negative": 1
   },
   {
    "default": "Stop",
@@ -276,7 +277,8 @@
    "description": "The percentage you are allowed to transfer more against the quantity ordered. For example, if you have ordered 100 units, and your Allowance is 10%, then you are allowed transfer 110 units.",
    "fieldname": "mr_qty_allowance",
    "fieldtype": "Float",
-   "label": "Over Transfer Allowance (%)"
+   "label": "Over Transfer Allowance (%)",
+   "non_negative": 1
   },
   {
    "default": "0",
@@ -437,7 +439,8 @@
    "description": "The percentage you are allowed to pick more items in the pick list than the ordered quantity.",
    "fieldname": "over_picking_allowance",
    "fieldtype": "Percent",
-   "label": "Over Picking Allowance (%)"
+   "label": "Over Picking Allowance (%)",
+   "non_negative": 1
   },
   {
    "default": "1",
@@ -590,7 +593,7 @@
  "index_web_pages_for_search": 1,
  "issingle": 1,
  "links": [],
- "modified": "2026-07-16 17:00:00.000000",
+ "modified": "2026-08-01 23:35:02.896836",
  "modified_by": "Administrator",
  "module": "Stock",
  "name": "Stock Settings",

erpnext/stock/doctype/stock_settings/stock_settings.py

@@ -68,7 +68,7 @@ class StockSettings(Document):
 		use_naming_series: DF.Check
 		use_serial_batch_fields: DF.Check
 		validate_material_transfer_warehouses: DF.Check
-		valuation_method: DF.Literal["FIFO", "Moving Average", "LIFO"]
+		valuation_method: DF.Literal["FIFO", "Moving Average", "LIFO", "Standard Cost"]
 	# end: auto-generated types
 
 	def validate(self):
@@ -101,6 +101,7 @@ class StockSettings(Document):
 				validate_fields_for_doctype=False,
 			)
 
+		self.validate_over_delivery_receipt_allowance()
 		self.validate_serial_and_batch_no_settings()
 		self.cant_change_valuation_method()
 		self.validate_clean_description_html()
@@ -112,6 +113,10 @@ class StockSettings(Document):
 		self.change_precision_for_stock_entry()
 		self.validate_do_not_use_batchwise_valuation()
 
+	def validate_over_delivery_receipt_allowance(self):
+		if not self.over_delivery_receipt_allowance:
+			self.role_allowed_to_over_deliver_receive = None
+
 	def validate_do_not_use_batchwise_valuation(self):
 		doc_before_save = self.get_doc_before_save()
 		if not doc_before_save:

erpnext/stock/get_item_details.py

@@ -40,6 +40,28 @@ purchase_doctypes = [
 
 NOT_APPLICABLE_TAX = "N/A"
 
+# For each transaction, the child-row link field(s) that point to the source
+# document item, mapped to that source item doctype. When "maintain same rate" is
+# on, a mapped row keeps the persisted source pricing (read straight from that row),
+# so an unsaved edit on the target row can never lock in a non-source rate.
+maintain_same_rate_source_fields = {
+	"Purchase Order": {"supplier_quotation_item": "Supplier Quotation Item"},
+	"Purchase Receipt": {"purchase_order_item": "Purchase Order Item"},
+	"Purchase Invoice": {"po_detail": "Purchase Order Item", "pr_detail": "Purchase Receipt Item"},
+	"Sales Order": {"quotation_item": "Quotation Item"},
+	"Delivery Note": {"so_detail": "Sales Order Item", "si_detail": "Sales Invoice Item"},
+	"Sales Invoice": {"so_detail": "Sales Order Item", "dn_detail": "Delivery Note Item"},
+}
+
+LOCKED_RATE_FIELDS = [
+	"price_list_rate",
+	"rate",
+	"discount_percentage",
+	"discount_amount",
+	"margin_type",
+	"margin_rate_or_amount",
+]
+
 
 def _preprocess_ctx(ctx):
 	if not ctx.price_list:
@@ -121,16 +143,20 @@ def get_item_details(
 	if ctx.doctype in ["Purchase Order", "Purchase Receipt", "Purchase Invoice"]:
 		ctx.customer = None
 
-	out.update(get_price_list_rate(ctx, item))
+	source_row = get_rate_locked_source_row(ctx, doc)
+	if source_row:
+		lock_source_rate(out, source_row)
+	else:
+		out.update(get_price_list_rate(ctx, item))
 
-	if (
-		not out.price_list_rate
-		and ctx.transaction_type == "selling"
-		and frappe.get_single_value("Selling Settings", "fallback_to_default_price_list")
-	):
-		fallback_args = ctx.copy()
-		fallback_args.price_list = frappe.get_single_value("Selling Settings", "selling_price_list")
-		out.update(get_price_list_rate(fallback_args, item))
+		if (
+			not out.price_list_rate
+			and ctx.transaction_type == "selling"
+			and frappe.get_single_value("Selling Settings", "fallback_to_default_price_list")
+		):
+			fallback_args = ctx.copy()
+			fallback_args.price_list = frappe.get_single_value("Selling Settings", "selling_price_list")
+			out.update(get_price_list_rate(fallback_args, item))
 
 	ctx.customer = current_customer
 
@@ -145,9 +171,8 @@ def get_item_details(
 		if ctx.get(key) is None:
 			ctx[key] = value
 
-	data = get_pricing_rule_for_item(ctx, doc=doc, for_validate=for_validate)
-
-	out.update(data)
+	if not source_row:
+		out.update(get_pricing_rule_for_item(ctx, doc=doc, for_validate=for_validate))
 
 	if (
 		frappe.get_single_value("Stock Settings", "auto_create_serial_and_batch_bundle_for_outward")
@@ -189,6 +214,61 @@ def remove_standard_fields(out: frappe._dict):
 	return out
 
 
+def get_rate_locked_source_row(ctx: ItemDetailsCtx, doc) -> frappe._dict | None:
+	"""Return the persisted source-document row a mapped target row is locked to.
+
+	The rate is read from the linked source row in the database (not the mutable
+	target row), so a re-fetch always restores the source pricing the maintain-same-
+	rate validator checks against, even after an unsaved edit on the target row.
+	"""
+	if isinstance(doc, str):
+		doc = json.loads(doc)
+
+	source_fields = maintain_same_rate_source_fields.get(ctx.parenttype or ctx.doctype)
+	if not source_fields or not doc or ctx.get("is_return") or not maintain_same_rate_enabled(ctx):
+		return None
+
+	row = next((d for d in doc.get("items") or [] if d.get("name") == ctx.child_docname), None)
+	if not row:
+		return None
+
+	for link_field, source_doctype in source_fields.items():
+		if source_name := row.get(link_field):
+			# a direct read would bypass permissions; only return source pricing to a
+			# caller allowed to read the source document
+			source = frappe.db.get_value(
+				source_doctype, source_name, [*LOCKED_RATE_FIELDS, "parent", "parenttype"], as_dict=True
+			)
+			if source and frappe.has_permission(source.parenttype, doc=source.parent):
+				return source
+			return None
+	return None
+
+
+def maintain_same_rate_enabled(ctx: ItemDetailsCtx) -> bool:
+	if (ctx.parenttype or ctx.doctype) in purchase_doctypes:
+		if ctx.get("is_internal_supplier"):
+			return False
+		return bool(cint(frappe.get_cached_value("Buying Settings", "None", "maintain_same_rate")))
+
+	if ctx.get("is_internal_customer"):
+		return False
+	return bool(cint(frappe.get_cached_value("Selling Settings", "None", "maintain_same_sales_rate")))
+
+
+def lock_source_rate(out: frappe._dict, source_row) -> None:
+	"""Copy the source row's whole pricing block onto out so a mapped row keeps its
+	exact rate. Pricing rules are skipped for these rows, so nothing re-derives it and
+	the manual discount or margin that made rate differ from price_list_rate survives.
+	"""
+	out.price_list_rate = flt(source_row.get("price_list_rate")) or flt(source_row.get("rate"))
+	out.rate = flt(source_row.get("rate"))
+	out.discount_percentage = flt(source_row.get("discount_percentage"))
+	out.discount_amount = flt(source_row.get("discount_amount"))
+	out.margin_type = source_row.get("margin_type")
+	out.margin_rate_or_amount = flt(source_row.get("margin_rate_or_amount"))
+
+
 def set_valuation_rate(out: frappe._dict, ctx: frappe._dict):
 	from erpnext.selling.doctype.product_bundle.product_bundle import get_active_product_bundle
 
@@ -1647,14 +1727,21 @@ def apply_price_list(ctx: ItemDetailsCtx, as_doc: bool = False, doc: Document |
 
 def apply_price_list_on_item(ctx, doc=None):
 	item_doc = frappe.get_cached_doc("Item", ctx.item_code)
-	item_details = get_price_list_rate(ctx, item_doc)
+
+	source_row = get_rate_locked_source_row(ctx, doc)
+	if source_row:
+		item_details = frappe._dict()
+		lock_source_rate(item_details, source_row)
+	else:
+		item_details = get_price_list_rate(ctx, item_doc)
 
 	ctx.conversion_factor = flt(ctx.conversion_factor) or get_conversion_factor(ctx.item_code, ctx.uom).get(
 		"conversion_factor", 1
 	)
 	ctx.stock_qty = flt(ctx.qty) * flt(ctx.conversion_factor)
 
-	item_details.update(get_pricing_rule_for_item(ctx, doc=doc))
+	if not source_row:
+		item_details.update(get_pricing_rule_for_item(ctx, doc=doc))
 
 	return item_details
 

erpnext/stock/report/stock_and_account_value_comparison/test_stock_and_account_value_comparison.py

@@ -8,6 +8,7 @@ from erpnext.stock.doctype.item.test_item import make_item
 from erpnext.stock.doctype.purchase_receipt.test_purchase_receipt import make_purchase_receipt
 from erpnext.stock.doctype.stock_entry.stock_entry_utils import make_stock_entry
 from erpnext.stock.doctype.warehouse.test_warehouse import create_warehouse
+from erpnext.stock.doctype.warehouse.warehouse import get_warehouses_based_on_account
 from erpnext.stock.report.stock_and_account_value_comparison.stock_and_account_value_comparison import (
 	create_reposting_entries,
 	execute,
@@ -113,6 +114,23 @@ class TestStockAndAccountValueComparison(ERPNextTestSuite):
 		)
 		self.assertFalse(item_wh_rivs, "Purchase vouchers must not be reposted Item-and-Warehouse based")
 
+	def test_child_account_override_excluded_from_group_account(self):
+		# A group warehouse carries an inventory account; a child (e.g. Goods-in-Transit) can override
+		# it with its own account. get_warehouses_based_on_account must return only warehouses whose
+		# effective account matches, excluding the overriding child.
+		group = create_warehouse("_Test SAVC Group WH", {"is_group": 1}, company=COMPANY)
+		group_account = frappe.get_value("Warehouse", group, "account")
+
+		inheriting = create_warehouse(
+			"_Test SAVC Inherit WH", {"parent_warehouse": group, "account": group_account}, company=COMPANY
+		)
+		overriding = create_warehouse("_Test SAVC Transit WH", {"parent_warehouse": group}, company=COMPANY)
+
+		warehouses = get_warehouses_based_on_account(group_account, COMPANY)
+
+		self.assertIn(inheriting, warehouses)
+		self.assertNotIn(overriding, warehouses)
+
 	def run_report(self, **extra):
 		filters = {"company": COMPANY, "as_on_date": "2026-12-31"}
 		filters.update(extra)

erpnext/stock/report/stock_ledger/stock_ledger.py

@@ -6,8 +6,10 @@ import copy
 
 import frappe
 from frappe import _
-from frappe.query_builder.functions import Sum
+from frappe.query_builder.functions import IfNull, Sum
 from frappe.utils import cint, flt, get_datetime
+from pypika import Order
+from pypika.analytics import RowNumber
 
 from erpnext.stock.doctype.inventory_dimension.inventory_dimension import get_inventory_dimensions
 from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos
@@ -52,14 +54,15 @@ def execute(filters=None):
 
 	data = []
 	conversion_factors = []
-	if opening_row:
-		data.append(opening_row)
+	opening_rows = opening_row if isinstance(opening_row, list) else ([opening_row] if opening_row else [])
+	for row in opening_rows:
+		data.append(row)
 		conversion_factors.append(0)
 
 	actual_qty = stock_value = 0
-	if opening_row:
-		actual_qty = opening_row.get("qty_after_transaction")
-		stock_value = opening_row.get("stock_value")
+	if opening_rows:
+		actual_qty = opening_rows[0].get("qty_after_transaction", 0)
+		stock_value = opening_rows[0].get("stock_value", 0)
 
 	available_serial_nos = {}
 
@@ -692,43 +695,120 @@ def get_opening_balance(filters, columns, sl_entries, inv_dimension_wise_value=N
 	if not (filters.item_code and filters.warehouse and filters.from_date):
 		return
 
-	from erpnext.stock.stock_ledger import get_previous_sle
+	item_codes = filters.item_code
+	if isinstance(item_codes, str):
+		item_codes = [item_codes]
 
-	project = None
-	if filters.get("project") and not frappe.get_all(
-		"Inventory Dimension", filters={"reference_document": "Project"}
-	):
-		project = filters.get("project")
+	warehouses = get_matching_warehouses(filters.warehouse)
+	if not warehouses:
+		return
 
-	last_entry = get_previous_sle(
-		{
-			"item_code": filters.item_code,
-			"warehouse_condition": get_warehouse_condition(filters.warehouse),
-			"posting_date": filters.from_date,
-			"posting_time": "00:00:00",
-			"project": project,
-		},
-		for_report=True,
+	sle_doctype = frappe.qb.DocType("Stock Ledger Entry")
+	sr_doctype = frappe.qb.DocType("Stock Reconciliation")
+
+	opening_reco_query = (
+		frappe.qb.from_(sle_doctype)
+		.inner_join(sr_doctype)
+		.on(sle_doctype.voucher_no == sr_doctype.name)
+		.select(sle_doctype.voucher_no)
+		.where(sle_doctype.docstatus < 2)
+		.where(sle_doctype.is_cancelled == 0)
+		.where(sle_doctype.item_code.isin(item_codes))
+		.where(sle_doctype.warehouse.isin(warehouses))
+		.where(sle_doctype.voucher_type == "Stock Reconciliation")
+		.where(sle_doctype.posting_date == filters.from_date)
+		.where(sr_doctype.purpose == "Opening Stock")
 	)
 
-	# check if any SLEs are actually Opening Stock Reconciliation
-	for sle in list(sl_entries):
-		if (
-			sle.get("voucher_type") == "Stock Reconciliation"
-			and sle.posting_date == filters.from_date
-			and frappe.db.get_value("Stock Reconciliation", sle.voucher_no, "purpose") == "Opening Stock"
-		):
-			last_entry = sle
-			sl_entries.remove(sle)
+	opening_reco_vouchers = set(opening_reco_query.run(pluck=True))
 
-	row = {
+	if opening_reco_vouchers:
+		sl_entries[:] = [sle for sle in sl_entries if sle.get("voucher_no") not in opening_reco_vouchers]
+
+	sle_cond = (sle_doctype.posting_date < filters.from_date) | (
+		(sle_doctype.posting_date == filters.from_date) & (sle_doctype.posting_time == "00:00:00")
+	)
+	if opening_reco_vouchers:
+		sle_cond = sle_cond | (
+			(sle_doctype.posting_date == filters.from_date)
+			& (sle_doctype.voucher_no.isin(list(opening_reco_vouchers)))
+		)
+
+	subq = (
+		frappe.qb.from_(sle_doctype)
+		.select(
+			sle_doctype.qty_after_transaction,
+			sle_doctype.stock_value,
+			RowNumber()
+			.over(sle_doctype.item_code, sle_doctype.warehouse)
+			.orderby(sle_doctype.posting_datetime, sle_doctype.creation, sle_doctype.name, order=Order.desc)
+			.as_("rn"),
+		)
+		.where(sle_doctype.docstatus < 2)
+		.where(sle_doctype.is_cancelled == 0)
+		.where(sle_doctype.item_code.isin(item_codes))
+		.where(sle_doctype.warehouse.isin(warehouses))
+		.where(sle_cond)
+	)
+
+	for field in ["voucher_no", "project", "company"]:
+		if filters.get(field):
+			subq = subq.where(sle_doctype[field] == filters.get(field))
+
+	inventory_dimension_fields = get_inventory_dimension_fields()
+	if inventory_dimension_fields:
+		for fieldname in inventory_dimension_fields:
+			if filters.get(fieldname):
+				subq = subq.where(sle_doctype[fieldname].isin(filters.get(fieldname)))
+
+	query = (
+		frappe.qb.from_(subq)
+		.select(
+			IfNull(Sum(subq.qty_after_transaction), 0.0).as_("total_qty"),
+			IfNull(Sum(subq.stock_value), 0.0).as_("total_stock_value"),
+		)
+		.where(subq.rn == 1)
+	)
+
+	res = query.run(as_dict=True)
+
+	total_qty = flt(res[0].total_qty) if res else 0.0
+	total_stock_value = flt(res[0].total_stock_value) if res else 0.0
+	valuation_rate = flt(total_stock_value / total_qty) if total_qty else 0.0
+
+	return {
 		"item_code": _("'Opening'"),
-		"qty_after_transaction": last_entry.get("qty_after_transaction", 0),
-		"valuation_rate": last_entry.get("valuation_rate", 0),
-		"stock_value": last_entry.get("stock_value", 0),
+		"qty_after_transaction": total_qty,
+		"valuation_rate": valuation_rate,
+		"stock_value": total_stock_value,
 	}
 
-	return row
+
+def get_matching_warehouses(warehouses):
+	if not warehouses:
+		return []
+
+	if isinstance(warehouses, str):
+		warehouses = [warehouses]
+
+	warehouse_details = frappe.get_all(
+		"Warehouse",
+		filters={"name": ("in", warehouses)},
+		fields=["lft", "rgt"],
+	)
+
+	if not warehouse_details:
+		return warehouses
+
+	wh = frappe.qb.DocType("Warehouse")
+	cond = None
+	for d in warehouse_details:
+		c = (wh.lft >= d.lft) & (wh.rgt <= d.rgt)
+		cond = c if cond is None else (cond | c)
+
+	matching = (frappe.qb.from_(wh).select(wh.name).where(cond)).run(pluck=True)
+
+	return matching if matching else warehouses
 
 
 def get_warehouse_condition(warehouses):
@@ -784,7 +864,15 @@ def get_opening_balance_for_inv_dimension(filters, inv_dimension_wise_value):
 	if not filters.item_code or not filters.warehouse or not filters.from_date:
 		return
 
-	if len(filters.get("item_code")) > 1 or len(filters.get("warehouse")) > 1:
+	item_codes = filters.get("item_code")
+	if isinstance(item_codes, str):
+		item_codes = [item_codes]
+
+	warehouses = filters.get("warehouse")
+	if isinstance(warehouses, str):
+		warehouses = [warehouses]
+
+	if len(item_codes) > 1 or len(warehouses) > 1:
 		return
 
 	sl_doctype = frappe.qb.DocType("Stock Ledger Entry")
@@ -804,17 +892,11 @@ def get_opening_balance_for_inv_dimension(filters, inv_dimension_wise_value):
 		)
 	)
 
-	if filters.get("item_code"):
-		if isinstance(filters.item_code, list | tuple):
-			query = query.where(sl_doctype.item_code.isin(filters.item_code))
-		else:
-			query = query.where(sl_doctype.item_code == filters.item_code)
+	if item_codes:
+		query = query.where(sl_doctype.item_code.isin(item_codes))
 
-	if filters.get("warehouse"):
-		if isinstance(filters.warehouse, list | tuple):
-			query = query.where(sl_doctype.warehouse.isin(filters.warehouse))
-		else:
-			query = query.where(sl_doctype.warehouse == filters.warehouse)
+	if warehouses:
+		query = query.where(sl_doctype.warehouse.isin(warehouses))
 
 	for key, value in inv_dimension_wise_value.items():
 		if isinstance(value, list | tuple):

erpnext/stock/report/stock_ledger/test_stock_ledger_report.py

@@ -87,3 +87,250 @@ class TestStockLedgerReport(ERPNextTestSuite):
 		rows = self.run_report(item_a)
 		item_codes = {row["item_code"] for row in rows if row.get("voucher_no")}
 		self.assertEqual(item_codes, {item_a})
+
+	def test_multi_item_opening_balance_with_and_without_transactions(self):
+		item_a = "_Test Item"
+		item_b = "_Test Item 2"
+		self.make_movements(
+			item_a,
+			[
+				{
+					"qty": 10,
+					"to_warehouse": WAREHOUSE,
+					"basic_rate": 100,
+					"posting_date": add_days(today(), -10),
+				}
+			],
+		)
+		self.make_movements(
+			item_b,
+			[{"qty": 5, "to_warehouse": WAREHOUSE, "basic_rate": 50, "posting_date": add_days(today(), -10)}],
+		)
+		self.make_movements(
+			item_a,
+			[{"qty": 2, "from_warehouse": WAREHOUSE, "posting_date": today()}],
+		)
+
+		filters = frappe._dict(
+			company="_Test Company",
+			from_date=add_days(today(), -5),
+			to_date=today(),
+			item_code=[item_a, item_b],
+			warehouse=WAREHOUSE,
+		)
+		columns, rows = execute(filters)
+
+		opening_rows = [row for row in rows if row.get("item_code") == "'Opening'"]
+		self.assertEqual(len(opening_rows), 1)
+		self.assertEqual(opening_rows[0]["qty_after_transaction"], 15)
+
+	def test_multi_warehouse_opening_balance_aggregation(self):
+		item = "_Test Item"
+		warehouse_1 = "Stores - _TC"
+		warehouse_2 = "Finished Goods - _TC"
+
+		self.make_movements(
+			item,
+			[
+				{
+					"qty": 10,
+					"to_warehouse": warehouse_1,
+					"basic_rate": 100,
+					"posting_date": add_days(today(), -10),
+				},
+				{
+					"qty": 20,
+					"to_warehouse": warehouse_2,
+					"basic_rate": 100,
+					"posting_date": add_days(today(), -10),
+				},
+			],
+		)
+
+		filters = frappe._dict(
+			company="_Test Company",
+			from_date=add_days(today(), -5),
+			to_date=today(),
+			item_code=[item],
+			warehouse=[warehouse_1, warehouse_2],
+		)
+		columns, rows = execute(filters)
+
+		opening_rows = [row for row in rows if row.get("item_code") == "'Opening'"]
+		self.assertEqual(len(opening_rows), 1)
+		self.assertEqual(opening_rows[0]["qty_after_transaction"], 30)
+
+	def test_opening_stock_reconciliation_on_from_date_non_midnight_time(self):
+		from erpnext.stock.doctype.stock_reconciliation.test_stock_reconciliation import (
+			create_stock_reconciliation,
+		)
+
+		item = "_Test Item"
+		from_date = today()
+
+		sr = create_stock_reconciliation(
+			item_code=item,
+			warehouse=WAREHOUSE,
+			qty=25,
+			rate=100,
+			posting_date=from_date,
+			posting_time="10:30:00",
+			purpose="Opening Stock",
+			do_not_submit=False,
+		)
+
+		filters = frappe._dict(
+			company="_Test Company",
+			from_date=from_date,
+			to_date=from_date,
+			item_code=[item],
+			warehouse=WAREHOUSE,
+		)
+		columns, rows = execute(filters)
+
+		opening_rows = [row for row in rows if row.get("item_code") == "'Opening'"]
+		self.assertEqual(len(opening_rows), 1)
+		self.assertEqual(opening_rows[0]["qty_after_transaction"], 25)
+
+		# Ensure the Opening Stock Reconciliation is not duplicated in detail transaction rows
+		reco_rows = [row for row in rows if row.get("voucher_no") == sr.name]
+		self.assertEqual(len(reco_rows), 0)
+
+	def test_backdated_sle_independent_maxima_handling(self):
+		item = "_Test Item"
+		# Entry 1: Later posting date (2026-07-20), created first
+		self.make_movements(
+			item,
+			[
+				{
+					"qty": 10,
+					"to_warehouse": WAREHOUSE,
+					"basic_rate": 100,
+					"posting_date": add_days(today(), -10),
+				}
+			],
+		)
+		# Entry 2: Backdated posting date (2026-07-15), created LATER
+		self.make_movements(
+			item,
+			[
+				{
+					"qty": 5,
+					"to_warehouse": WAREHOUSE,
+					"basic_rate": 100,
+					"posting_date": add_days(today(), -15),
+				}
+			],
+		)
+
+		filters = frappe._dict(
+			company="_Test Company",
+			from_date=add_days(today(), -5),
+			to_date=today(),
+			item_code=[item],
+			warehouse=WAREHOUSE,
+		)
+		columns, rows = execute(filters)
+
+		opening_rows = [row for row in rows if row.get("item_code") == "'Opening'"]
+		self.assertEqual(len(opening_rows), 1)
+		# Should correctly pick the latest posting date entry (15 Qty) despite backdated creation order
+		self.assertEqual(opening_rows[0]["qty_after_transaction"], 15)
+
+	def test_filtered_opening_balance_does_not_pick_excluded_creation_entry(self):
+		item = "_Test Item"
+		posting_date = add_days(today(), -10)
+		posting_time = "09:00:00"
+
+		included_entry = make_stock_entry(
+			item_code=item,
+			qty=10,
+			to_warehouse=WAREHOUSE,
+			basic_rate=100,
+			posting_date=posting_date,
+			posting_time=posting_time,
+		)
+		make_stock_entry(
+			item_code=item,
+			qty=50,
+			to_warehouse=WAREHOUSE,
+			basic_rate=100,
+			posting_date=posting_date,
+			posting_time=posting_time,
+		)
+
+		filters = frappe._dict(
+			company="_Test Company",
+			from_date=add_days(today(), -5),
+			to_date=today(),
+			item_code=[item],
+			warehouse=WAREHOUSE,
+			voucher_no=included_entry.name,
+		)
+		columns, rows = execute(filters)
+
+		opening_rows = [row for row in rows if row.get("item_code") == "'Opening'"]
+		self.assertEqual(len(opening_rows), 1)
+		self.assertEqual(opening_rows[0]["qty_after_transaction"], 10)
+
+	def test_tied_creation_terminal_sle_is_not_summed_twice(self):
+		item = "_Test Item"
+		posting_date = add_days(today(), -10)
+		posting_time = "09:00:00"
+
+		stock_entry_1 = make_stock_entry(
+			item_code=item,
+			qty=10,
+			to_warehouse=WAREHOUSE,
+			basic_rate=100,
+			posting_date=posting_date,
+			posting_time=posting_time,
+		)
+		stock_entry_2 = make_stock_entry(
+			item_code=item,
+			qty=5,
+			to_warehouse=WAREHOUSE,
+			basic_rate=100,
+			posting_date=posting_date,
+			posting_time=posting_time,
+		)
+
+		sle_rows = frappe.get_all(
+			"Stock Ledger Entry",
+			filters={
+				"voucher_type": "Stock Entry",
+				"voucher_no": ("in", [stock_entry_1.name, stock_entry_2.name]),
+				"item_code": item,
+				"warehouse": WAREHOUSE,
+				"is_cancelled": 0,
+			},
+			fields=["name", "qty_after_transaction"],
+			order_by="name desc",
+		)
+		self.assertEqual(len(sle_rows), 2)
+
+		for sle in sle_rows:
+			frappe.db.set_value(
+				"Stock Ledger Entry",
+				sle.name,
+				"creation",
+				"2026-01-01 00:00:00.000000",
+				update_modified=False,
+			)
+
+		filters = frappe._dict(
+			company="_Test Company",
+			from_date=add_days(today(), -5),
+			to_date=today(),
+			item_code=[item],
+			warehouse=WAREHOUSE,
+		)
+		columns, rows = execute(filters)
+
+		opening_rows = [row for row in rows if row.get("item_code") == "'Opening'"]
+		self.assertEqual(len(opening_rows), 1)
+		self.assertEqual(opening_rows[0]["qty_after_transaction"], sle_rows[0].qty_after_transaction)
+		self.assertNotEqual(
+			opening_rows[0]["qty_after_transaction"],
+			sum(sle.qty_after_transaction for sle in sle_rows),
+		)

erpnext/stock/serial_batch_bundle.py

@@ -6,6 +6,7 @@ from frappe.model.naming import NamingSeries, parse_naming_series
 from frappe.query_builder.functions import Max, Sum
 from frappe.utils import add_days, cint, cstr, flt, get_link_to_form, getdate, now
 from pypika import Order
+from pypika.terms import ExistsCriterion
 
 from erpnext.stock.deprecated_serial_batch import (
 	DeprecatedBatchNoValuation,
@@ -830,13 +831,14 @@ class BatchNoValuation(DeprecatedBatchNoValuation):
 					("batch-valuation", self.sle.item_code, self.sle.warehouse)
 				)
 
-			entries = self.get_batch_stock_before_date()
 			self.stock_value_change = 0.0
 			self.batch_avg_rate = defaultdict(float)
 			self.available_qty = defaultdict(float)
 			self.stock_value_differece = defaultdict(float)
 
-			for ledger in entries:
+			self.seed_from_stock_closing_balance()
+
+			for ledger in self.get_batch_stock_before_date():
 				self.stock_value_differece[ledger.batch_no] += flt(ledger.incoming_rate)
 				self.available_qty[ledger.batch_no] += flt(ledger.qty)
 
@@ -844,6 +846,52 @@ class BatchNoValuation(DeprecatedBatchNoValuation):
 			self.calculate_avg_rate_for_non_batchwise_valuation()
 			self.set_stock_value_difference()
 
+	def seed_from_stock_closing_balance(self):
+		self.stock_closing_from_datetime = None
+		closing_entry = self.get_closing_entry_for_seeding()
+		if not closing_entry:
+			return
+
+		from erpnext.stock.utils import get_combine_datetime
+
+		self.stock_closing_from_datetime = get_combine_datetime(
+			add_days(closing_entry.to_date, 1), "00:00:00"
+		)
+
+		for row in self.get_stock_closing_balance_entries(closing_entry.name):
+			self.stock_value_differece[row.batch_no] += flt(row.stock_value_difference)
+			self.available_qty[row.batch_no] += flt(row.actual_qty)
+
+	def get_closing_entry_for_seeding(self):
+		from erpnext.stock.doctype.stock_closing_entry.stock_closing_entry import (
+			get_closing_entry_for_closed_period,
+		)
+
+		if not self.batchwise_valuation_batches or not self.sle.posting_date:
+			return None
+
+		company = self.sle.company or frappe.get_cached_value("Warehouse", self.sle.warehouse, "company")
+		closing_entry = get_closing_entry_for_closed_period(company)
+		if not closing_entry or getdate(self.sle.posting_date) <= getdate(closing_entry.to_date):
+			return None
+
+		return closing_entry
+
+	def get_stock_closing_balance_entries(self, closing_entry):
+		table = frappe.qb.DocType("Stock Closing Balance")
+
+		return (
+			frappe.qb.from_(table)
+			.select(table.batch_no, table.actual_qty, table.stock_value_difference)
+			.where(
+				(table.stock_closing_entry == closing_entry)
+				& (table.item_code == self.sle.item_code)
+				& (table.warehouse == self.sle.warehouse)
+				& table.batch_no.isin(self.batchwise_valuation_batches)
+				& (table.inventory_dimension_key.isnull() | (table.inventory_dimension_key == ""))
+			)
+		).run(as_dict=True)
+
 	def get_batch_stock_before_date(self) -> list[dict]:
 		# Get batch wise stock value difference from Serial and Batch Bundle considering time condition
 		if not self.batchwise_valuation_batches:
@@ -851,14 +899,45 @@ class BatchNoValuation(DeprecatedBatchNoValuation):
 
 		child = frappe.qb.DocType("Serial and Batch Entry")
 
+		sle_creation = self.sle.creation if self.sle.get("name") else None
+		if not self.sle.get("name") and self.sle.get("serial_and_batch_bundle"):
+			sle_creation = frappe.db.get_value(
+				"Stock Ledger Entry",
+				{"serial_and_batch_bundle": self.sle.serial_and_batch_bundle, "is_cancelled": 0},
+				"creation",
+			)
+
 		timestamp_condition = ""
 		if self.sle.posting_datetime:
 			timestamp_condition = child.posting_datetime < self.sle.posting_datetime
 
-			if self.sle.creation:
-				timestamp_condition |= (child.posting_datetime == self.sle.posting_datetime) & (
-					child.creation < self.sle.creation
+			sle_table = frappe.qb.DocType("Stock Ledger Entry")
+			if sle_creation:
+				# bundle creation and SLE creation are different timelines (a
+				# bundle can be created much before its SLE), so break the tie
+				# using the creation of the bundle's own SLE
+				tie_condition = ExistsCriterion(
+					frappe.qb.from_(sle_table)
+					.select(sle_table.name)
+					.where(
+						(sle_table.serial_and_batch_bundle == child.parent)
+						& (sle_table.is_cancelled == 0)
+						& (sle_table.creation < sle_creation)
+					)
 				)
+			else:
+				# the current entry is not yet in the ledger and will get the
+				# latest creation, so the same-timestamp entries which are
+				# already in the ledger precede it
+				tie_condition = ExistsCriterion(
+					frappe.qb.from_(sle_table)
+					.select(sle_table.name)
+					.where(
+						(sle_table.serial_and_batch_bundle == child.parent) & (sle_table.is_cancelled == 0)
+					)
+				)
+
+			timestamp_condition |= (child.posting_datetime == self.sle.posting_datetime) & tie_condition
 
 		conditions = (
 			(child.item_code == self.sle.item_code)
@@ -878,6 +957,9 @@ class BatchNoValuation(DeprecatedBatchNoValuation):
 		if timestamp_condition:
 			conditions &= timestamp_condition
 
+		if self.stock_closing_from_datetime:
+			conditions &= child.posting_datetime >= self.stock_closing_from_datetime
+
 		# MariaDB carries a row lock on the grouped query below; on postgres the caller
 		# (calculate_avg_rate) serializes via a txn-scoped advisory lock on (item, warehouse)
 		# instead of row-locking the whole history (FOR UPDATE is invalid with GROUP BY there).

erpnext/stock/services/quality_inspection_service.py

@@ -50,9 +50,25 @@ QI_OUTGOING_PURPOSES = (
 )
 
 
+SECONDARY_ITEM_PURPOSES = ("Manufacture", "Repack", "Disassemble")
+
+
+def is_inspection_exempt_secondary_row(doc, row) -> bool:
+	"""Whether the row is a secondary item on a document that produces secondary items."""
+	if not (row.get("secondary_item_type") or row.get("is_legacy_scrap_item")):
+		return False
+
+	if doc.doctype == "Stock Entry":
+		return doc.purpose in SECONDARY_ITEM_PURPOSES
+
+	return True
+
+
 def stock_entry_row_requires_inspection(purpose, row):
 	"""Check if this Stock Entry row need a Quality Inspection."""
-	if row.get("secondary_item_type") or row.get("is_legacy_scrap_item"):
+	if purpose in SECONDARY_ITEM_PURPOSES and (
+		row.get("secondary_item_type") or row.get("is_legacy_scrap_item")
+	):
 		return False
 	if purpose == "Manufacture":
 		return bool(row.is_finished_item)
@@ -88,7 +104,7 @@ class QualityInspectionService:
 			elif self.doc.doctype == "Stock Entry":
 				qi_required = stock_entry_row_requires_inspection(self.doc.purpose, row)
 
-			if row.get("secondary_item_type") or row.get("is_legacy_scrap_item"):
+			if is_inspection_exempt_secondary_row(self.doc, row):
 				continue
 
 			if qi_required:  # validate row only if inspection is required on item level

erpnext/stock/stock_ledger.py

@@ -101,6 +101,32 @@ def validate_standard_cost_posting_date(sl_entries):
 			)
 
 
+def validate_stock_frozen_by_closing_entry(sl_entries):
+	from erpnext.stock.doctype.stock_closing_entry.stock_closing_entry import (
+		get_closing_entry_for_closed_period,
+	)
+
+	company = sl_entries[0].get("company")
+	if not company:
+		company = frappe.get_cached_value("Warehouse", sl_entries[0].get("warehouse"), "company")
+
+	closing_entry = get_closing_entry_for_closed_period(company)
+	if not closing_entry:
+		return
+
+	for sle in sl_entries:
+		if sle.get("posting_date") and getdate(sle.get("posting_date")) <= getdate(closing_entry.to_date):
+			frappe.throw(
+				_(
+					"Stock transactions dated on or before {0} are frozen because the period is closed and the Stock Closing Entry {1} has been generated. To make changes, cancel the Period Closing Voucher first."
+				).format(
+					frappe.bold(format_date(closing_entry.to_date)),
+					get_link_to_form("Stock Closing Entry", closing_entry.name),
+				),
+				title=_("Stock Frozen"),
+			)
+
+
 def make_sl_entries(sl_entries, allow_negative_stock=False, via_landed_cost_voucher=False):
 	"""Create SL entries from SL entry dicts
 
@@ -119,6 +145,8 @@ def make_sl_entries(sl_entries, allow_negative_stock=False, via_landed_cost_vouc
 		for pair in sorted({(d.get("item_code"), d.get("warehouse")) for d in sl_entries}):
 			sle_processing_gate(*pair)
 
+		validate_stock_frozen_by_closing_entry(sl_entries)
+
 		cancelled = sl_entries[0].get("is_cancelled")
 		if cancelled:
 			validate_cancellation(sl_entries)

erpnext/stock/tests/test_get_item_details.py

@@ -124,3 +124,336 @@ class TestGetItemDetail(ERPNextTestSuite):
 		dn.save()
 		self.assertEqual(dn.items[0].batch_no, "BATCH01")
 		self.assertEqual(dn.items[0].rate, 50)
+
+	def test_maintain_same_rate_keeps_source_rate_on_refetch(self):
+		"""#57436: with "maintain same rate" on, re-fetching a PR row mapped from a
+		PO must keep the PO rate instead of pulling a newer, higher Item Price.
+
+		The rate is validated on save, so it can never persist changed; assert the
+		fetched rate directly to prove the newer Item Price is never picked up.
+		"""
+		from erpnext.buying.doctype.purchase_order.mapper import make_purchase_receipt
+		from erpnext.buying.doctype.purchase_order.test_purchase_order import create_purchase_order
+		from erpnext.stock.doctype.item.test_item import make_item
+
+		def set_maintain_same_rate(value):
+			frappe.db.set_single_value("Buying Settings", "maintain_same_rate", value)
+			frappe.clear_cache(doctype="Buying Settings")
+
+		set_maintain_same_rate(1)
+
+		item_code = make_item(properties={"is_stock_item": 1}).name
+		po = create_purchase_order(item_code=item_code, qty=1, rate=100)
+
+		# The PO may auto-insert an Item Price at 100; bump it to the newer, higher rate.
+		item_price = frappe.db.get_value(
+			"Item Price", {"item_code": item_code, "price_list": "Standard Buying"}
+		)
+		if item_price:
+			frappe.db.set_value("Item Price", item_price, "price_list_rate", 120)
+		else:
+			frappe.get_doc(
+				{
+					"doctype": "Item Price",
+					"price_list": "Standard Buying",
+					"item_code": item_code,
+					"price_list_rate": 120,
+				}
+			).insert()
+
+		pr = make_purchase_receipt(po.name)
+		pr.insert()
+
+		def fetch_price_list_rate():
+			ctx = frappe._dict(
+				{
+					"item_code": item_code,
+					"doctype": "Purchase Receipt",
+					"name": pr.name,
+					"company": pr.company,
+					"supplier": pr.supplier,
+					"currency": pr.currency,
+					"conversion_rate": 1.0,
+					"price_list": "Standard Buying",
+					"price_list_currency": pr.currency,
+					"plc_conversion_rate": 1.0,
+					"warehouse": pr.items[0].warehouse,
+					"uom": pr.items[0].uom,
+					"stock_uom": pr.items[0].stock_uom,
+					"qty": pr.items[0].qty,
+					"child_doctype": pr.items[0].doctype,
+					"child_docname": pr.items[0].name,
+					"is_return": 0,
+					"is_internal_supplier": 0,
+					"ignore_pricing_rule": 1,
+				}
+			)
+			return get_item_details(ctx, pr).get("price_list_rate")
+
+		# Rate stays at the PO rate; the newer Item Price (120) is not fetched.
+		self.assertEqual(fetch_price_list_rate(), 100)
+
+		# Control: without the setting the newer Item Price would be fetched.
+		set_maintain_same_rate(0)
+		self.assertEqual(fetch_price_list_rate(), 120)
+
+	def test_maintain_same_rate_survives_refetch_with_discount(self):
+		"""A mapped Purchase Receipt row that carries a source discount (rate != price
+		list rate) must keep its rate when the row is re-fetched, so maintain-same-rate
+		lets the document save. process_item_selection runs the same recompute the desk
+		mirrors, so it covers the "discount discarded on refresh" concern end to end.
+		"""
+		from frappe.utils import flt
+
+		from erpnext.buying.doctype.purchase_order.mapper import make_purchase_receipt
+		from erpnext.buying.doctype.purchase_order.test_purchase_order import create_purchase_order
+
+		item, price_list = "_Test Item", "_Test Buying Price List"
+		original = frappe.db.get_single_value("Buying Settings", "maintain_same_rate")
+		original_action = frappe.db.get_single_value("Buying Settings", "maintain_same_rate_action")
+		frappe.db.set_single_value("Buying Settings", "maintain_same_rate", 1)
+		frappe.db.set_single_value("Buying Settings", "maintain_same_rate_action", "Stop")
+		frappe.clear_cache(doctype="Buying Settings")
+
+		try:
+			for label, adjustment in (
+				("percentage", {"discount_percentage": 10}),
+				("amount", {"discount_amount": 10}),
+			):
+				with self.subTest(discount=label):
+					# a controlled discounted PO: list rate 100, effective rate 90
+					frappe.flags.dont_fetch_price_list_rate = True
+					po = create_purchase_order(item_code=item, qty=1, do_not_save=True)
+					po.buying_price_list = price_list
+					po.items[0].price_list_rate = 100
+					po.items[0].update(adjustment)
+					po.items[0].rate = 90
+					po.insert()
+					po.submit()
+					frappe.flags.dont_fetch_price_list_rate = False
+
+					# a newer Item Price must not leak onto the mapped row on re-fetch
+					item_price = frappe.db.get_value(
+						"Item Price", {"item_code": item, "price_list": price_list}
+					)
+					if item_price:
+						frappe.db.set_value("Item Price", item_price, "price_list_rate", 250)
+
+					pr = make_purchase_receipt(po.name)
+					pr.insert()
+					pr.process_item_selection(item_idx=pr.items[0].idx)
+
+					self.assertEqual(flt(pr.items[0].rate), 90)
+					pr.save()  # must not raise the maintain-same-rate check
+		finally:
+			frappe.db.set_single_value("Buying Settings", "maintain_same_rate", original)
+			frappe.db.set_single_value("Buying Settings", "maintain_same_rate_action", original_action)
+			frappe.clear_cache(doctype="Buying Settings")
+			frappe.flags.dont_fetch_price_list_rate = False
+
+	def test_apply_price_list_keeps_source_rate_when_maintain_same_rate(self):
+		"""#57436: the bulk apply_price_list path (price list / party / conversion rate
+		change) must also keep the source rate on mapped rows, not just re-fetch of a
+		single row. Here a PR row carries its PO rate (175) while the current price list
+		rate is 100; the bulk apply must keep 175.
+		"""
+		from frappe.utils import flt, nowdate
+
+		from erpnext.buying.doctype.purchase_order.test_purchase_order import create_purchase_order
+		from erpnext.stock.get_item_details import apply_price_list
+
+		item_code = "_Test Item"
+		price_list = "_Test Buying Price List"
+
+		original = frappe.db.get_single_value("Buying Settings", "maintain_same_rate")
+		frappe.db.set_single_value("Buying Settings", "maintain_same_rate", 1)
+		frappe.clear_cache(doctype="Buying Settings")
+
+		try:
+			po = create_purchase_order(item_code=item_code, rate=175, qty=1)
+
+			row_name = "pr-row-1"
+			pr_doc = {
+				"doctype": "Purchase Receipt",
+				"items": [
+					{
+						"name": row_name,
+						"item_code": item_code,
+						"purchase_order_item": po.items[0].name,
+						"price_list_rate": 175,
+						"rate": 175,
+					}
+				],
+			}
+			ctx = frappe._dict(
+				doctype="Purchase Receipt",
+				supplier=po.supplier,
+				company=po.company,
+				currency=po.currency,
+				conversion_rate=1.0,
+				price_list=price_list,
+				plc_conversion_rate=1.0,
+				transaction_date=nowdate(),
+				items=[
+					frappe._dict(
+						doctype="Purchase Receipt Item",
+						parenttype="Purchase Receipt",
+						item_code=item_code,
+						child_docname=row_name,
+						qty=1,
+						uom=po.items[0].uom,
+						stock_uom=po.items[0].stock_uom,
+						conversion_factor=1.0,
+					)
+				],
+			)
+
+			result = apply_price_list(ctx, doc=pr_doc)
+			self.assertEqual(flt(result["children"][0].get("price_list_rate")), 175)
+		finally:
+			frappe.db.set_single_value("Buying Settings", "maintain_same_rate", original)
+			frappe.clear_cache(doctype="Buying Settings")
+
+	def test_maintain_same_rate_keeps_source_discount_on_refetch(self):
+		"""A mapped source row with a discount has rate != price_list_rate. Re-fetch must
+		return the source's rate and discount, not just the pre-discount price, or the
+		recomputed rate diverges from the reference and fails maintain-same-rate on save.
+		"""
+		from frappe.utils import flt
+
+		from erpnext.buying.doctype.purchase_order.test_purchase_order import create_purchase_order
+
+		item_code = "_Test Item"
+		price_list = "_Test Buying Price List"
+
+		original = frappe.db.get_single_value("Buying Settings", "maintain_same_rate")
+		frappe.db.set_single_value("Buying Settings", "maintain_same_rate", 1)
+		frappe.clear_cache(doctype="Buying Settings")
+
+		try:
+			# source PO carries the discount: list rate 100, 10% off, effective rate 90
+			frappe.flags.dont_fetch_price_list_rate = True
+			po = create_purchase_order(item_code=item_code, qty=1, do_not_save=True)
+			po.buying_price_list = price_list
+			po.items[0].price_list_rate = 100
+			po.items[0].discount_percentage = 10
+			po.items[0].rate = 90
+			po.insert()
+			po.submit()
+			frappe.flags.dont_fetch_price_list_rate = False
+
+			row_name = "pr-row-1"
+			pr_doc = {
+				"doctype": "Purchase Receipt",
+				"items": [
+					{"name": row_name, "item_code": item_code, "purchase_order_item": po.items[0].name}
+				],
+			}
+			ctx = frappe._dict(
+				item_code=item_code,
+				doctype="Purchase Receipt",
+				company=po.company,
+				supplier=po.supplier,
+				currency=po.currency,
+				conversion_rate=1.0,
+				price_list=price_list,
+				price_list_currency=po.currency,
+				plc_conversion_rate=1.0,
+				warehouse="_Test Warehouse - _TC",
+				uom=po.items[0].uom,
+				stock_uom=po.items[0].stock_uom,
+				qty=1,
+				child_docname=row_name,
+				is_return=0,
+				is_internal_supplier=0,
+				ignore_pricing_rule=1,
+			)
+
+			out = get_item_details(ctx, pr_doc)
+			self.assertEqual(flt(out.get("price_list_rate")), 100)
+			self.assertEqual(flt(out.get("rate")), 90)
+			self.assertEqual(flt(out.get("discount_percentage")), 10)
+		finally:
+			frappe.db.set_single_value("Buying Settings", "maintain_same_rate", original)
+			frappe.clear_cache(doctype="Buying Settings")
+			frappe.flags.dont_fetch_price_list_rate = False
+
+	def test_refetch_restores_source_rate_after_target_edit(self):
+		"""Editing a mapped row's rate then re-fetching must restore the persisted source
+		rate (read from the linked row), not lock in the edit, so the document still saves.
+		"""
+		from frappe.utils import flt
+
+		from erpnext.buying.doctype.purchase_order.mapper import make_purchase_receipt
+		from erpnext.buying.doctype.purchase_order.test_purchase_order import create_purchase_order
+
+		item = "_Test Item"
+		original = frappe.db.get_single_value("Buying Settings", "maintain_same_rate")
+		original_action = frappe.db.get_single_value("Buying Settings", "maintain_same_rate_action")
+		frappe.db.set_single_value("Buying Settings", "maintain_same_rate", 1)
+		frappe.db.set_single_value("Buying Settings", "maintain_same_rate_action", "Stop")
+		frappe.clear_cache(doctype="Buying Settings")
+
+		try:
+			po = create_purchase_order(item_code=item, qty=1, rate=90)
+			pr = make_purchase_receipt(po.name)
+			pr.insert()
+
+			# user edits the mapped row to a non-source rate
+			pr.items[0].price_list_rate = 200
+			pr.items[0].rate = 200
+
+			# a re-fetch must restore the persisted source (PO) rate, not keep the edit
+			pr.process_item_selection(item_idx=pr.items[0].idx)
+			self.assertEqual(flt(pr.items[0].rate), 90)
+			pr.save()  # must not raise the maintain-same-rate check
+		finally:
+			frappe.db.set_single_value("Buying Settings", "maintain_same_rate", original)
+			frappe.db.set_single_value("Buying Settings", "maintain_same_rate_action", original_action)
+			frappe.clear_cache(doctype="Buying Settings")
+
+	def test_rate_lock_source_lookup_checks_permission(self):
+		"""The lock reads source pricing via a direct DB read, so it must not disclose a
+		source document's pricing to a caller who cannot read that document.
+		"""
+		from erpnext.buying.doctype.purchase_order.test_purchase_order import create_purchase_order
+		from erpnext.stock.get_item_details import get_rate_locked_source_row
+
+		original = frappe.db.get_single_value("Buying Settings", "maintain_same_rate")
+		frappe.db.set_single_value("Buying Settings", "maintain_same_rate", 1)
+		frappe.clear_cache(doctype="Buying Settings")
+
+		role, email = "_Test Role Without PO Access", "_test_rate_lock_probe@example.com"
+		try:
+			po = create_purchase_order(item_code="_Test Item", qty=1, rate=90)
+			pr_doc = {
+				"doctype": "Purchase Receipt",
+				"items": [{"name": "r1", "item_code": "_Test Item", "purchase_order_item": po.items[0].name}],
+			}
+			ctx = frappe._dict(doctype="Purchase Receipt", child_docname="r1")
+
+			# an authorized caller receives the source row
+			self.assertIsNotNone(get_rate_locked_source_row(ctx.copy(), dict(pr_doc)))
+
+			if not frappe.db.exists("Role", role):
+				frappe.get_doc({"doctype": "Role", "role_name": role, "desk_access": 1}).insert(
+					ignore_permissions=True
+				)
+			if not frappe.db.exists("User", email):
+				frappe.get_doc(
+					{
+						"doctype": "User",
+						"email": email,
+						"first_name": "Probe",
+						"send_welcome_email": 0,
+						"roles": [{"role": role}],
+					}
+				).insert(ignore_permissions=True)
+
+			frappe.set_user(email)
+			# a caller who cannot read the Purchase Order gets nothing
+			self.assertIsNone(get_rate_locked_source_row(ctx.copy(), dict(pr_doc)))
+		finally:
+			frappe.set_user("Administrator")
+			frappe.db.set_single_value("Buying Settings", "maintain_same_rate", original)
+			frappe.clear_cache(doctype="Buying Settings")

erpnext/utilities/transaction_base.py

@@ -547,7 +547,9 @@ class TransactionBase(StatusUpdater):
 		from erpnext.stock.get_item_details import apply_price_list
 
 		args = {
-			"items": [x.as_dict() for x in self.items],
+			# pass child_docname so the maintain-same-rate lock in apply_price_list can
+			# match each row, consistent with the desk (JS) callers
+			"items": [{**x.as_dict(), "child_docname": x.name} for x in self.items],
 			"customer": self.customer or self.party_name,
 			"quotation_to": self.quotation_to,
 			"customer_group": self.customer_group,