diff --git a/erpnext/accounts/doctype/pos_invoice/pos_invoice.py b/erpnext/accounts/doctype/pos_invoice/pos_invoice.py
index 0ce2c1a4f25..9e6c518dfc0 100644
--- a/erpnext/accounts/doctype/pos_invoice/pos_invoice.py
+++ b/erpnext/accounts/doctype/pos_invoice/pos_invoice.py
@@ -228,6 +228,7 @@ class POSInvoice(SalesInvoice):
 			self.apply_loyalty_points()
 		self.check_phone_payments()
 		self.set_status(update=True)
+		self.make_bundle_for_sales_purchase_return()
 		self.submit_serial_batch_bundle()
 
 		if self.coupon_code:
diff --git a/erpnext/accounts/doctype/pos_invoice/test_pos_invoice.py b/erpnext/accounts/doctype/pos_invoice/test_pos_invoice.py
index 773ef0103b6..9159f837269 100644
--- a/erpnext/accounts/doctype/pos_invoice/test_pos_invoice.py
+++ b/erpnext/accounts/doctype/pos_invoice/test_pos_invoice.py
@@ -318,29 +318,28 @@ class TestPOSInvoice(unittest.TestCase):
 
 		pos.insert()
 		pos.submit()
+		pos.reload()
 
 		pos_return1 = make_sales_return(pos.name)
 
 		# partial return 1
 		pos_return1.get("items")[0].qty = -1
+		pos_return1.submit()
+		pos_return1.reload()
 
 		bundle_id = frappe.get_doc(
 			"Serial and Batch Bundle", pos_return1.get("items")[0].serial_and_batch_bundle
 		)
 
-		bundle_id.remove(bundle_id.entries[1])
-		bundle_id.save()
-
 		bundle_id.load_from_db()
 
 		serial_no = bundle_id.entries[0].serial_no
 		self.assertEqual(serial_no, serial_nos[0])
 
-		pos_return1.insert()
-		pos_return1.submit()
-
 		# partial return 2
 		pos_return2 = make_sales_return(pos.name)
+		pos_return2.submit()
+
 		self.assertEqual(pos_return2.get("items")[0].qty, -1)
 		serial_no = get_serial_nos_from_bundle(pos_return2.get("items")[0].serial_and_batch_bundle)[0]
 		self.assertEqual(serial_no, serial_nos[1])
diff --git a/erpnext/accounts/doctype/purchase_invoice/purchase_invoice.py b/erpnext/accounts/doctype/purchase_invoice/purchase_invoice.py
index cebd730e489..bb158bedcd0 100644
--- a/erpnext/accounts/doctype/purchase_invoice/purchase_invoice.py
+++ b/erpnext/accounts/doctype/purchase_invoice/purchase_invoice.py
@@ -708,6 +708,7 @@ class PurchaseInvoice(BuyingController):
 		# Updating stock ledger should always be called after updating prevdoc status,
 		# because updating ordered qty in bin depends upon updated ordered qty in PO
 		if self.update_stock == 1:
+			self.make_bundle_for_sales_purchase_return()
 			self.make_bundle_using_old_serial_batch_fields()
 			self.update_stock_ledger()
 
diff --git a/erpnext/accounts/doctype/sales_invoice/sales_invoice.py b/erpnext/accounts/doctype/sales_invoice/sales_invoice.py
index ba79cfee663..f588d14b43a 100644
--- a/erpnext/accounts/doctype/sales_invoice/sales_invoice.py
+++ b/erpnext/accounts/doctype/sales_invoice/sales_invoice.py
@@ -455,6 +455,7 @@ class SalesInvoice(SellingController):
 				if not self.get(table_name):
 					continue
 
+				self.make_bundle_for_sales_purchase_return(table_name)
 				self.make_bundle_using_old_serial_batch_fields(table_name)
 			self.update_stock_ledger()
 
diff --git a/erpnext/controllers/sales_and_purchase_return.py b/erpnext/controllers/sales_and_purchase_return.py
index 83f3410d8fb..f60759e9ee8 100644
--- a/erpnext/controllers/sales_and_purchase_return.py
+++ b/erpnext/controllers/sales_and_purchase_return.py
@@ -1,11 +1,12 @@
 # Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
 # License: GNU General Public License v3. See license.txt
 
+from collections import defaultdict
 
 import frappe
 from frappe import _
 from frappe.model.meta import get_field_precision
-from frappe.utils import flt, format_datetime, get_datetime
+from frappe.utils import cint, flt, format_datetime, get_datetime
 
 import erpnext
 from erpnext.stock.serial_batch_bundle import get_batches_from_bundle
@@ -513,6 +514,7 @@ def make_return_doc(doctype: str, source_name: str, target_doc=None, return_agai
 				target_doc.rejected_warehouse = ""
 				target_doc.warehouse = source_doc.rejected_warehouse
 				target_doc.received_qty = target_doc.qty
+				target_doc.return_qty_from_rejected_warehouse = 1
 
 		elif doctype == "Purchase Invoice":
 			returned_qty_map = get_returned_qty_map_for_row(
@@ -570,7 +572,14 @@ def make_return_doc(doctype: str, source_name: str, target_doc=None, return_agai
 			if default_warehouse_for_sales_return:
 				target_doc.warehouse = default_warehouse_for_sales_return
 
-		if source_doc.item_code:
+		if (
+			(source_doc.serial_no or source_doc.batch_no)
+			and not source_doc.serial_and_batch_bundle
+			and not source_doc.use_serial_batch_fields
+		):
+			target_doc.set("use_serial_batch_fields", 1)
+
+		if source_doc.item_code and target_doc.get("use_serial_batch_fields"):
 			item_details = frappe.get_cached_value(
 				"Item", source_doc.item_code, ["has_batch_no", "has_serial_no"], as_dict=1
 			)
@@ -578,14 +587,7 @@ def make_return_doc(doctype: str, source_name: str, target_doc=None, return_agai
 			if not item_details.has_batch_no and not item_details.has_serial_no:
 				return
 
-			if not target_doc.get("use_serial_batch_fields"):
-				for qty_field in ["stock_qty", "rejected_qty"]:
-					if not target_doc.get(qty_field):
-						continue
-
-					update_serial_batch_no(source_doc, target_doc, source_parent, item_details, qty_field)
-			elif target_doc.get("use_serial_batch_fields"):
-				update_non_bundled_serial_nos(source_doc, target_doc, source_parent)
+			update_non_bundled_serial_nos(source_doc, target_doc, source_parent)
 
 	def update_non_bundled_serial_nos(source_doc, target_doc, source_parent):
 		from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos
@@ -839,3 +841,229 @@ def get_returned_batches(child_doc, parent_doc, batch_no_field=None, ignore_vouc
 		batches.update(get_batches_from_bundle(ids))
 
 	return batches
+
+
+def available_serial_batch_for_return(field, doctype, reference_ids, is_rejected=False):
+	available_dict = get_available_serial_batches(field, doctype, reference_ids, is_rejected=is_rejected)
+	if not available_dict:
+		frappe.throw(_("No Serial / Batches are available for return"))
+
+	return available_dict
+
+
+def get_available_serial_batches(field, doctype, reference_ids, is_rejected=False):
+	_bundle_ids = get_serial_and_batch_bundle(field, doctype, reference_ids, is_rejected=is_rejected)
+	if not _bundle_ids:
+		return frappe._dict({})
+
+	return get_serial_batches_based_on_bundle(field, _bundle_ids)
+
+
+def get_serial_batches_based_on_bundle(field, _bundle_ids):
+	available_dict = frappe._dict({})
+	batch_serial_nos = frappe.get_all(
+		"Serial and Batch Bundle",
+		fields=[
+			"`tabSerial and Batch Entry`.`serial_no`",
+			"`tabSerial and Batch Entry`.`batch_no`",
+			"`tabSerial and Batch Entry`.`qty`",
+			"`tabSerial and Batch Bundle`.`voucher_detail_no`",
+			"`tabSerial and Batch Bundle`.`voucher_type`",
+			"`tabSerial and Batch Bundle`.`voucher_no`",
+		],
+		filters=[
+			["Serial and Batch Bundle", "name", "in", _bundle_ids],
+			["Serial and Batch Entry", "docstatus", "=", 1],
+		],
+		order_by="`tabSerial and Batch Bundle`.`creation`, `tabSerial and Batch Entry`.`idx`",
+	)
+
+	for row in batch_serial_nos:
+		key = row.voucher_detail_no
+		if frappe.get_cached_value(row.voucher_type, row.voucher_no, "is_return"):
+			key = frappe.get_cached_value(row.voucher_type + " Item", row.voucher_detail_no, field)
+
+		if key not in available_dict:
+			available_dict[key] = frappe._dict(
+				{"qty": 0.0, "serial_nos": defaultdict(float), "batches": defaultdict(float)}
+			)
+
+		available_dict[key]["qty"] += row.qty
+
+		if row.serial_no:
+			available_dict[key]["serial_nos"][row.serial_no] += row.qty
+		elif row.batch_no:
+			available_dict[key]["batches"][row.batch_no] += row.qty
+
+	return available_dict
+
+
+def get_serial_and_batch_bundle(field, doctype, reference_ids, is_rejected=False):
+	filters = {"docstatus": 1, "name": ("in", reference_ids), "serial_and_batch_bundle": ("is", "set")}
+
+	pluck_field = "serial_and_batch_bundle"
+	if is_rejected:
+		del filters["serial_and_batch_bundle"]
+		filters["rejected_serial_and_batch_bundle"] = ("is", "set")
+		pluck_field = "rejected_serial_and_batch_bundle"
+
+	_bundle_ids = frappe.get_all(
+		doctype,
+		filters=filters,
+		pluck=pluck_field,
+	)
+
+	if not _bundle_ids:
+		return {}
+
+	del filters["name"]
+
+	filters[field] = ("in", reference_ids)
+
+	if not is_rejected:
+		_bundle_ids.extend(
+			frappe.get_all(
+				doctype,
+				filters=filters,
+				pluck="serial_and_batch_bundle",
+			)
+		)
+	else:
+		fields = [
+			"serial_and_batch_bundle",
+		]
+
+		if is_rejected:
+			fields.extend(["rejected_serial_and_batch_bundle", "return_qty_from_rejected_warehouse"])
+
+		data = frappe.get_all(
+			doctype,
+			fields=fields,
+			filters=filters,
+		)
+
+		for d in data:
+			if is_rejected:
+				if d.get("return_qty_from_rejected_warehouse"):
+					_bundle_ids.append(d.get("serial_and_batch_bundle"))
+				else:
+					_bundle_ids.append(d.get("rejected_serial_and_batch_bundle"))
+			else:
+				_bundle_ids.append(d.get("serial_and_batch_bundle"))
+
+	return _bundle_ids
+
+
+def filter_serial_batches(parent_doc, data, row, warehouse_field=None, qty_field=None):
+	if not qty_field:
+		qty_field = "qty"
+
+	if not warehouse_field:
+		warehouse_field = "warehouse"
+
+	warehouse = row.get(warehouse_field)
+	qty = abs(row.get(qty_field))
+
+	filterd_serial_batch = frappe._dict({"serial_nos": [], "batches": defaultdict(float)})
+
+	if data.serial_nos:
+		available_serial_nos = []
+		for serial_no, sn_qty in data.serial_nos.items():
+			if sn_qty != 0:
+				available_serial_nos.append(serial_no)
+
+		if available_serial_nos:
+			if parent_doc.doctype in ["Purchase Invoice", "Purchase Reecipt"]:
+				available_serial_nos = get_available_serial_nos(available_serial_nos)
+
+			if len(available_serial_nos) > qty:
+				filterd_serial_batch["serial_nos"] = sorted(available_serial_nos[0 : cint(qty)])
+			else:
+				filterd_serial_batch["serial_nos"] = available_serial_nos
+
+	elif data.batches:
+		for batch_no, batch_qty in data.batches.items():
+			if parent_doc.get("is_internal_customer"):
+				batch_qty = batch_qty * -1
+
+			if batch_qty <= 0:
+				continue
+
+			if parent_doc.doctype in ["Purchase Invoice", "Purchase Reecipt"]:
+				batch_qty = get_available_batch_qty(
+					parent_doc,
+					batch_no,
+					warehouse,
+				)
+
+				if batch_qty <= 0:
+					frappe.throw(
+						_("Batch {0} is not available in warehouse {1}").format(batch_no, warehouse),
+						title=_("Batch Not Available for Return"),
+					)
+
+			if qty <= 0:
+				break
+
+			if batch_qty > qty:
+				filterd_serial_batch["batches"][batch_no] = qty
+				qty = 0
+			else:
+				filterd_serial_batch["batches"][batch_no] += batch_qty
+				qty -= batch_qty
+
+	return filterd_serial_batch
+
+
+def get_available_batch_qty(parent_doc, batch_no, warehouse):
+	from erpnext.stock.doctype.batch.batch import get_batch_qty
+
+	return get_batch_qty(
+		batch_no,
+		warehouse,
+		posting_date=parent_doc.posting_date,
+		posting_time=parent_doc.posting_time,
+		for_stock_levels=True,
+	)
+
+
+def make_serial_batch_bundle_for_return(data, child_doc, parent_doc, warehouse_field=None):
+	from erpnext.stock.serial_batch_bundle import SerialBatchCreation
+
+	type_of_transaction = "Outward"
+	if parent_doc.doctype in ["Sales Invoice", "Delivery Note", "POS Invoice"]:
+		type_of_transaction = "Inward"
+
+	if not warehouse_field:
+		warehouse_field = "warehouse"
+
+	warehouse = child_doc.get(warehouse_field)
+	if parent_doc.get("is_internal_customer"):
+		warehouse = child_doc.get("target_warehouse")
+		type_of_transaction = "Outward"
+
+	cls_obj = SerialBatchCreation(
+		{
+			"type_of_transaction": type_of_transaction,
+			"item_code": child_doc.item_code,
+			"warehouse": warehouse,
+			"serial_nos": data.get("serial_nos"),
+			"batches": data.get("batches"),
+			"posting_date": parent_doc.posting_date,
+			"posting_time": parent_doc.posting_time,
+			"voucher_type": parent_doc.doctype,
+			"voucher_no": parent_doc.name,
+			"voucher_detail_no": child_doc.name,
+			"qty": child_doc.qty,
+			"company": parent_doc.company,
+			"do_not_submit": True,
+		}
+	).make_serial_and_batch_bundle()
+
+	return cls_obj.name
+
+
+def get_available_serial_nos(serial_nos, warehouse):
+	return frappe.get_all(
+		"Serial No", filters={"warehouse": warehouse, "name": ("in", serial_nos)}, pluck="name"
+	)
diff --git a/erpnext/controllers/stock_controller.py b/erpnext/controllers/stock_controller.py
index 524bbfcc4b1..7ced1eb1c54 100644
--- a/erpnext/controllers/stock_controller.py
+++ b/erpnext/controllers/stock_controller.py
@@ -16,6 +16,11 @@ from erpnext.accounts.general_ledger import (
 )
 from erpnext.accounts.utils import cancel_exchange_gain_loss_journal, get_fiscal_year
 from erpnext.controllers.accounts_controller import AccountsController
+from erpnext.controllers.sales_and_purchase_return import (
+	available_serial_batch_for_return,
+	filter_serial_batches,
+	make_serial_batch_bundle_for_return,
+)
 from erpnext.stock import get_warehouse_account_map
 from erpnext.stock.doctype.inventory_dimension.inventory_dimension import (
 	get_evaluated_inventory_dimension,
@@ -217,6 +222,125 @@ class StockController(AccountsController):
 					self.update_bundle_details(bundle_details, table_name, row, is_rejected=True)
 					self.create_serial_batch_bundle(bundle_details, row)
 
+	def make_bundle_for_sales_purchase_return(self, table_name=None):
+		if not self.get("is_return"):
+			return
+
+		if not table_name:
+			table_name = "items"
+
+		self.make_bundle_for_non_rejected_qty(table_name)
+
+		if self.doctype in ["Purchase Invoice", "Purchase Receipt"]:
+			self.make_bundle_for_rejected_qty(table_name)
+
+	def make_bundle_for_rejected_qty(self, table_name=None):
+		field, reference_ids = self.get_reference_ids(
+			table_name, "rejected_qty", "rejected_serial_and_batch_bundle"
+		)
+
+		if not reference_ids:
+			return
+
+		child_doctype = self.doctype + " Item"
+		available_dict = available_serial_batch_for_return(
+			field, child_doctype, reference_ids, is_rejected=True
+		)
+
+		for row in self.get(table_name):
+			if data := available_dict.get(row.get(field)):
+				qty_field = "rejected_qty"
+				warehouse_field = "rejected_warehouse"
+				if row.get("return_qty_from_rejected_warehouse"):
+					qty_field = "qty"
+					warehouse_field = "warehouse"
+
+				data = filter_serial_batches(
+					self, data, row, warehouse_field=warehouse_field, qty_field=qty_field
+				)
+				bundle = make_serial_batch_bundle_for_return(data, row, self, warehouse_field)
+				if row.get("return_qty_from_rejected_warehouse"):
+					row.db_set(
+						{
+							"serial_and_batch_bundle": bundle,
+							"batch_no": "",
+							"serial_no": "",
+						}
+					)
+				else:
+					row.db_set(
+						{
+							"rejected_serial_and_batch_bundle": bundle,
+							"batch_no": "",
+							"rejected_serial_no": "",
+						}
+					)
+
+	def make_bundle_for_non_rejected_qty(self, table_name):
+		field, reference_ids = self.get_reference_ids(table_name)
+		if not reference_ids:
+			return
+
+		child_doctype = self.doctype + " Item"
+		available_dict = available_serial_batch_for_return(field, child_doctype, reference_ids)
+
+		for row in self.get(table_name):
+			if data := available_dict.get(row.get(field)):
+				data = filter_serial_batches(self, data, row)
+				bundle = make_serial_batch_bundle_for_return(data, row, self)
+				row.db_set(
+					{
+						"serial_and_batch_bundle": bundle,
+						"batch_no": "",
+						"serial_no": "",
+					}
+				)
+
+	def get_reference_ids(self, table_name, qty_field=None, bundle_field=None) -> tuple[str, list[str]]:
+		field = {
+			"Sales Invoice": "sales_invoice_item",
+			"Delivery Note": "dn_detail",
+			"Purchase Receipt": "purchase_receipt_item",
+			"Purchase Invoice": "purchase_invoice_item",
+			"POS Invoice": "pos_invoice_item",
+		}.get(self.doctype)
+
+		if not bundle_field:
+			bundle_field = "serial_and_batch_bundle"
+
+		if not qty_field:
+			qty_field = "qty"
+
+		reference_ids = []
+
+		for row in self.get(table_name):
+			if not self.is_serial_batch_item(row.item_code):
+				continue
+
+			if (
+				row.get(field)
+				and (
+					qty_field == "qty"
+					and not row.get("return_qty_from_rejected_warehouse")
+					or qty_field == "rejected_qty"
+					and (row.get("return_qty_from_rejected_warehouse") or row.get("rejected_warehouse"))
+				)
+				and not row.get("use_serial_batch_fields")
+				and not row.get(bundle_field)
+			):
+				reference_ids.append(row.get(field))
+
+		return field, reference_ids
+
+	@frappe.request_cache
+	def is_serial_batch_item(self, item_code) -> bool:
+		item_details = frappe.db.get_value("Item", item_code, ["has_serial_no", "has_batch_no"], as_dict=1)
+
+		if item_details.has_serial_no or item_details.has_batch_no:
+			return True
+
+		return False
+
 	def update_bundle_details(self, bundle_details, table_name, row, is_rejected=False):
 		from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos
 
@@ -611,35 +735,16 @@ class StockController(AccountsController):
 	def make_package_for_transfer(
 		self, serial_and_batch_bundle, warehouse, type_of_transaction=None, do_not_submit=None
 	):
-		bundle_doc = frappe.get_doc("Serial and Batch Bundle", serial_and_batch_bundle)
-
-		if not type_of_transaction:
-			type_of_transaction = "Inward"
-
-		bundle_doc = frappe.copy_doc(bundle_doc)
-		bundle_doc.warehouse = warehouse
-		bundle_doc.type_of_transaction = type_of_transaction
-		bundle_doc.voucher_type = self.doctype
-		bundle_doc.voucher_no = "" if self.is_new() or self.docstatus == 2 else self.name
-		bundle_doc.is_cancelled = 0
-
-		for row in bundle_doc.entries:
-			row.is_outward = 0
-			row.qty = abs(row.qty)
-			row.stock_value_difference = abs(row.stock_value_difference)
-			if type_of_transaction == "Outward":
-				row.qty *= -1
-				row.stock_value_difference *= row.stock_value_difference
-				row.is_outward = 1
-
-			row.warehouse = warehouse
-
-		bundle_doc.calculate_qty_and_amount()
-		bundle_doc.flags.ignore_permissions = True
-		bundle_doc.flags.ignore_validate = True
-		bundle_doc.save(ignore_permissions=True)
-
-		return bundle_doc.name
+		return make_bundle_for_material_transfer(
+			is_new=self.is_new(),
+			docstatus=self.docstatus,
+			voucher_type=self.doctype,
+			voucher_no=self.name,
+			serial_and_batch_bundle=serial_and_batch_bundle,
+			warehouse=warehouse,
+			type_of_transaction=type_of_transaction,
+			do_not_submit=do_not_submit,
+		)
 
 	def get_sl_entries(self, d, args):
 		sl_dict = frappe._dict(
@@ -1557,3 +1662,38 @@ def create_item_wise_repost_entries(
 		repost_entries.append(repost_entry)
 
 	return repost_entries
+
+
+def make_bundle_for_material_transfer(**kwargs):
+	if isinstance(kwargs, dict):
+		kwargs = frappe._dict(kwargs)
+
+	bundle_doc = frappe.get_doc("Serial and Batch Bundle", kwargs.serial_and_batch_bundle)
+
+	if not kwargs.type_of_transaction:
+		kwargs.type_of_transaction = "Inward"
+
+	bundle_doc = frappe.copy_doc(bundle_doc)
+	bundle_doc.warehouse = kwargs.warehouse
+	bundle_doc.type_of_transaction = kwargs.type_of_transaction
+	bundle_doc.voucher_type = kwargs.voucher_type
+	bundle_doc.voucher_no = "" if kwargs.is_new or kwargs.docstatus == 2 else kwargs.voucher_no
+	bundle_doc.is_cancelled = 0
+
+	for row in bundle_doc.entries:
+		row.is_outward = 0
+		row.qty = abs(row.qty)
+		row.stock_value_difference = abs(row.stock_value_difference)
+		if kwargs.type_of_transaction == "Outward":
+			row.qty *= -1
+			row.stock_value_difference *= row.stock_value_difference
+			row.is_outward = 1
+
+		row.warehouse = kwargs.warehouse
+
+	bundle_doc.calculate_qty_and_amount()
+	bundle_doc.flags.ignore_permissions = True
+	bundle_doc.flags.ignore_validate = True
+	bundle_doc.save(ignore_permissions=True)
+
+	return bundle_doc.name
diff --git a/erpnext/stock/deprecated_serial_batch.py b/erpnext/stock/deprecated_serial_batch.py
index f2a53858a9a..79ace06d25a 100644
--- a/erpnext/stock/deprecated_serial_batch.py
+++ b/erpnext/stock/deprecated_serial_batch.py
@@ -8,8 +8,11 @@ from pypika import Order
 class DeprecatedSerialNoValuation:
 	@deprecated
 	def calculate_stock_value_from_deprecarated_ledgers(self):
-		if not frappe.db.get_value(
-			"Stock Ledger Entry", {"serial_no": ("is", "set"), "is_cancelled": 0}, "name"
+		if not frappe.db.get_all(
+			"Stock Ledger Entry",
+			fields=["name"],
+			filters={"serial_no": ("is", "set"), "is_cancelled": 0, "item_code": self.sle.item_code},
+			limit=1,
 		):
 			return
 
@@ -41,6 +44,12 @@ class DeprecatedSerialNoValuation:
 		# get rate from serial nos within same company
 		incoming_values = 0.0
 		for serial_no in serial_nos:
+			sn_details = frappe.db.get_value("Serial No", serial_no, ["purchase_rate", "company"], as_dict=1)
+			if sn_details and sn_details.purchase_rate and sn_details.company == self.sle.company:
+				self.serial_no_incoming_rate[serial_no] += flt(sn_details.purchase_rate)
+				incoming_values += self.serial_no_incoming_rate[serial_no]
+				continue
+
 			table = frappe.qb.DocType("Stock Ledger Entry")
 			stock_ledgers = (
 				frappe.qb.from_(table)
diff --git a/erpnext/stock/doctype/batch/batch.py b/erpnext/stock/doctype/batch/batch.py
index 77b87aa995c..0be85e46015 100644
--- a/erpnext/stock/doctype/batch/batch.py
+++ b/erpnext/stock/doctype/batch/batch.py
@@ -208,7 +208,8 @@ def get_batch_qty(
 
 	:param batch_no: Optional - give qty for this batch no
 	:param warehouse: Optional - give qty for this warehouse
-	:param item_code: Optional - give qty for this item"""
+	:param item_code: Optional - give qty for this item
+	:param for_stock_levels: True consider expired batches"""
 
 	from erpnext.stock.doctype.serial_and_batch_bundle.serial_and_batch_bundle import (
 		get_auto_batch_nos,
diff --git a/erpnext/stock/doctype/delivery_note/delivery_note.py b/erpnext/stock/doctype/delivery_note/delivery_note.py
index 013415e1cc7..df6536624fa 100644
--- a/erpnext/stock/doctype/delivery_note/delivery_note.py
+++ b/erpnext/stock/doctype/delivery_note/delivery_note.py
@@ -468,6 +468,7 @@ class DeliveryNote(SellingController):
 			if not self.get(table_name):
 				continue
 
+			self.make_bundle_for_sales_purchase_return(table_name)
 			self.make_bundle_using_old_serial_batch_fields(table_name)
 
 		# Updating stock ledger should always be called after updating prevdoc status,
@@ -1365,6 +1366,9 @@ def make_inter_company_transaction(doctype, source_name, target_doc=None):
 		if source_parent.doctype == "Delivery Note" and source.received_qty:
 			target.qty = flt(source.qty) + flt(source.returned_qty) - flt(source.received_qty)
 
+		if source.get("use_serial_batch_fields"):
+			target.set("use_serial_batch_fields", 1)
+
 	doclist = get_mapped_doc(
 		doctype,
 		source_name,
diff --git a/erpnext/stock/doctype/delivery_note/test_delivery_note.py b/erpnext/stock/doctype/delivery_note/test_delivery_note.py
index e01a3397b64..58ea0be5d77 100644
--- a/erpnext/stock/doctype/delivery_note/test_delivery_note.py
+++ b/erpnext/stock/doctype/delivery_note/test_delivery_note.py
@@ -261,18 +261,15 @@ class TestDeliveryNote(FrappeTestCase):
 		self.assertTrue(dn.items[0].serial_no)
 
 		frappe.flags.ignore_serial_batch_bundle_validation = False
+		frappe.flags.use_serial_and_batch_fields = False
 
 		# return entry
 		dn1 = make_sales_return(dn.name)
 
 		dn1.items[0].qty = -2
-
-		bundle_doc = frappe.get_doc("Serial and Batch Bundle", dn1.items[0].serial_and_batch_bundle)
-		bundle_doc.set("entries", bundle_doc.entries[:2])
-		bundle_doc.save()
-
-		dn1.save()
+		dn1.items[0].serial_no = "\n".join(get_serial_nos(serial_nos)[0:2])
 		dn1.submit()
+		dn1.reload()
 
 		returned_serial_nos1 = get_serial_nos_from_bundle(dn1.items[0].serial_and_batch_bundle)
 		for serial_no in returned_serial_nos1:
@@ -281,21 +278,15 @@ class TestDeliveryNote(FrappeTestCase):
 		dn2 = make_sales_return(dn.name)
 
 		dn2.items[0].qty = -2
-
-		bundle_doc = frappe.get_doc("Serial and Batch Bundle", dn2.items[0].serial_and_batch_bundle)
-		bundle_doc.set("entries", bundle_doc.entries[:2])
-		bundle_doc.save()
-
-		dn2.save()
+		dn2.items[0].serial_no = "\n".join(get_serial_nos(serial_nos)[2:4])
 		dn2.submit()
+		dn2.reload()
 
 		returned_serial_nos2 = get_serial_nos_from_bundle(dn2.items[0].serial_and_batch_bundle)
 		for serial_no in returned_serial_nos2:
 			self.assertTrue(serial_no in serial_nos)
 			self.assertFalse(serial_no in returned_serial_nos1)
 
-		frappe.flags.use_serial_and_batch_fields = False
-
 	def test_sales_return_for_non_bundled_items_partial(self):
 		company = frappe.db.get_value("Warehouse", "Stores - TCP1", "company")
 
@@ -440,7 +431,7 @@ class TestDeliveryNote(FrappeTestCase):
 		self.assertEqual(dn.per_returned, 100)
 		self.assertEqual(dn.status, "Return Issued")
 
-	def test_delivery_note_return_valuation_on_different_warehuose(self):
+	def test_delivery_note_return_valuation_on_different_warehouse(self):
 		from erpnext.stock.doctype.warehouse.test_warehouse import create_warehouse
 
 		company = frappe.db.get_value("Warehouse", "Stores - TCP1", "company")
diff --git a/erpnext/stock/doctype/purchase_receipt/purchase_receipt.py b/erpnext/stock/doctype/purchase_receipt/purchase_receipt.py
index 94ad468cbc9..1146e0aab4f 100644
--- a/erpnext/stock/doctype/purchase_receipt/purchase_receipt.py
+++ b/erpnext/stock/doctype/purchase_receipt/purchase_receipt.py
@@ -371,6 +371,7 @@ class PurchaseReceipt(BuyingController):
 		else:
 			self.db_set("status", "Completed")
 
+		self.make_bundle_for_sales_purchase_return()
 		self.make_bundle_using_old_serial_batch_fields()
 		# Updating stock ledger should always be called after updating prevdoc status,
 		# because updating ordered qty, reserved_qty_for_subcontract in bin
diff --git a/erpnext/stock/doctype/purchase_receipt/test_purchase_receipt.py b/erpnext/stock/doctype/purchase_receipt/test_purchase_receipt.py
index 830a5e99567..4d1e8e841e9 100644
--- a/erpnext/stock/doctype/purchase_receipt/test_purchase_receipt.py
+++ b/erpnext/stock/doctype/purchase_receipt/test_purchase_receipt.py
@@ -2685,7 +2685,7 @@ class TestPurchaseReceipt(FrappeTestCase):
 		for row in inter_transfer_dn_return.items:
 			self.assertTrue(row.serial_and_batch_bundle)
 
-	def test_internal_transfer_with_serial_batch_items_without_user_serial_batch_fields(self):
+	def test_internal_transfer_with_serial_batch_items_without_use_serial_batch_fields(self):
 		from erpnext.controllers.sales_and_purchase_return import make_return_doc
 		from erpnext.stock.doctype.delivery_note.delivery_note import make_inter_company_purchase_receipt
 		from erpnext.stock.doctype.delivery_note.test_delivery_note import create_delivery_note
diff --git a/erpnext/stock/doctype/purchase_receipt_item/purchase_receipt_item.json b/erpnext/stock/doctype/purchase_receipt_item/purchase_receipt_item.json
index 6ba1469a46a..9604c55450e 100644
--- a/erpnext/stock/doctype/purchase_receipt_item/purchase_receipt_item.json
+++ b/erpnext/stock/doctype/purchase_receipt_item/purchase_receipt_item.json
@@ -81,6 +81,7 @@
   "purchase_invoice",
   "column_break_40",
   "allow_zero_valuation_rate",
+  "return_qty_from_rejected_warehouse",
   "is_fixed_asset",
   "asset_location",
   "asset_category",
@@ -1116,12 +1117,19 @@
    "hidden": 1,
    "label": "Apply TDS",
    "read_only": 1
+  },
+  {
+   "default": "0",
+   "fieldname": "return_qty_from_rejected_warehouse",
+   "fieldtype": "Check",
+   "label": "Return Qty from Rejected Warehouse",
+   "read_only": 1
   }
  ],
  "idx": 1,
  "istable": 1,
  "links": [],
- "modified": "2024-04-08 20:00:16.277292",
+ "modified": "2024-05-28 09:48:24.448815",
  "modified_by": "Administrator",
  "module": "Stock",
  "name": "Purchase Receipt Item",
diff --git a/erpnext/stock/doctype/purchase_receipt_item/purchase_receipt_item.py b/erpnext/stock/doctype/purchase_receipt_item/purchase_receipt_item.py
index 908c0a7a0f4..393b6a25691 100644
--- a/erpnext/stock/doctype/purchase_receipt_item/purchase_receipt_item.py
+++ b/erpnext/stock/doctype/purchase_receipt_item/purchase_receipt_item.py
@@ -85,6 +85,7 @@ class PurchaseReceiptItem(Document):
 		rejected_serial_no: DF.Text | None
 		rejected_warehouse: DF.Link | None
 		retain_sample: DF.Check
+		return_qty_from_rejected_warehouse: DF.Check
 		returned_qty: DF.Float
 		rm_supp_cost: DF.Currency
 		sales_order: DF.Link | None
diff --git a/erpnext/stock/doctype/putaway_rule/test_putaway_rule.py b/erpnext/stock/doctype/putaway_rule/test_putaway_rule.py
index 9f7fdeccd05..9178229c018 100644
--- a/erpnext/stock/doctype/putaway_rule/test_putaway_rule.py
+++ b/erpnext/stock/doctype/putaway_rule/test_putaway_rule.py
@@ -377,7 +377,7 @@ class TestPutawayRule(FrappeTestCase):
 			apply_putaway_rule=1,
 			do_not_save=1,
 		)
-		stock_entry.save()
+		stock_entry.submit()
 		stock_entry.load_from_db()
 
 		self.assertEqual(stock_entry.items[0].t_warehouse, self.warehouse_1)
@@ -398,11 +398,17 @@ class TestPutawayRule(FrappeTestCase):
 
 		self.assertUnchangedItemsOnResave(stock_entry)
 
-		for row in stock_entry.items:
-			if row.serial_and_batch_bundle:
-				frappe.delete_doc("Serial and Batch Bundle", row.serial_and_batch_bundle)
-
 		stock_entry.load_from_db()
+		stock_entry.cancel()
+
+		rivs = frappe.get_all("Repost Item Valuation", filters={"voucher_no": stock_entry.name})
+		for row in rivs:
+			riv_doc = frappe.get_doc("Repost Item Valuation", row.name)
+			riv_doc.cancel()
+			riv_doc.delete()
+
+		frappe.db.set_single_value("Accounts Settings", "delete_linked_ledger_entries", 1)
+
 		stock_entry.delete()
 		pr.cancel()
 		rule_1.delete()
diff --git a/erpnext/stock/doctype/serial_and_batch_bundle/serial_and_batch_bundle.py b/erpnext/stock/doctype/serial_and_batch_bundle/serial_and_batch_bundle.py
index 5e16115db01..ad757351a9b 100644
--- a/erpnext/stock/doctype/serial_and_batch_bundle/serial_and_batch_bundle.py
+++ b/erpnext/stock/doctype/serial_and_batch_bundle/serial_and_batch_bundle.py
@@ -156,6 +156,8 @@ class SerialandBatchBundle(Document):
 
 	def validate_serial_nos_duplicate(self):
 		# Don't inward same serial number multiple times
+		if self.voucher_type in ["POS Invoice", "Pick List"]:
+			return
 
 		if not self.warehouse:
 			return
diff --git a/erpnext/stock/doctype/stock_entry/stock_entry.js b/erpnext/stock/doctype/stock_entry/stock_entry.js
index fb63f1c23c6..4e00de0d7ce 100644
--- a/erpnext/stock/doctype/stock_entry/stock_entry.js
+++ b/erpnext/stock/doctype/stock_entry/stock_entry.js
@@ -111,6 +111,8 @@ frappe.ui.form.on("Stock Entry", {
 				// or a pre-existing batch
 				if (frm.doc.purpose != "Material Receipt") {
 					filters["warehouse"] = item.s_warehouse || item.t_warehouse;
+				} else {
+					filters["is_inward"] = 1;
 				}
 
 				return {
@@ -1110,6 +1112,13 @@ erpnext.stock.StockEntry = class StockEntry extends erpnext.stock.StockControlle
 
 	on_submit() {
 		this.clean_up();
+		this.refresh_serial_batch_bundle_field();
+	}
+
+	refresh_serial_batch_bundle_field() {
+		frappe.route_hooks.after_submit = (frm_obj) => {
+			frm_obj.reload_doc();
+		};
 	}
 
 	after_cancel() {
diff --git a/erpnext/stock/doctype/stock_entry/stock_entry.py b/erpnext/stock/doctype/stock_entry/stock_entry.py
index 9b66fb2ba95..61be2849326 100644
--- a/erpnext/stock/doctype/stock_entry/stock_entry.py
+++ b/erpnext/stock/doctype/stock_entry/stock_entry.py
@@ -226,6 +226,7 @@ class StockEntry(StockController):
 		if not self.from_bom:
 			self.fg_completed_qty = 0.0
 
+		self.make_serial_and_batch_bundle_for_outward()
 		self.validate_serialized_batch()
 		self.calculate_rate_and_amount()
 		self.validate_putaway_capacity()
@@ -289,9 +290,6 @@ class StockEntry(StockController):
 		if self.purpose == "Material Transfer" and self.outgoing_stock_entry:
 			self.set_material_request_transfer_status("In Transit")
 
-	def before_save(self):
-		self.make_serial_and_batch_bundle_for_outward()
-
 	def on_update(self):
 		self.set_serial_and_batch_bundle()
 
@@ -997,7 +995,7 @@ class StockEntry(StockController):
 			self.purpose = frappe.get_cached_value("Stock Entry Type", self.stock_entry_type, "purpose")
 
 	def make_serial_and_batch_bundle_for_outward(self):
-		if self.docstatus == 1:
+		if self.docstatus == 0:
 			return
 
 		serial_or_batch_items = get_serial_or_batch_items(self.items)
@@ -1050,12 +1048,11 @@ class StockEntry(StockController):
 			if not bundle_doc:
 				continue
 
-			if self.docstatus == 0:
-				for entry in bundle_doc.entries:
-					if not entry.serial_no:
-						continue
+			for entry in bundle_doc.entries:
+				if not entry.serial_no:
+					continue
 
-					already_picked_serial_nos.append(entry.serial_no)
+				already_picked_serial_nos.append(entry.serial_no)
 
 			row.serial_and_batch_bundle = bundle_doc.name
 
diff --git a/erpnext/stock/report/batch_wise_balance_history/batch_wise_balance_history.py b/erpnext/stock/report/batch_wise_balance_history/batch_wise_balance_history.py
index 16a0de57a5d..822da13cc72 100644
--- a/erpnext/stock/report/batch_wise_balance_history/batch_wise_balance_history.py
+++ b/erpnext/stock/report/batch_wise_balance_history/batch_wise_balance_history.py
@@ -4,7 +4,7 @@
 
 import frappe
 from frappe import _
-from frappe.utils import cint, flt, get_table_name, getdate
+from frappe.utils import add_to_date, cint, flt, get_datetime, get_table_name, getdate
 from frappe.utils.deprecations import deprecated
 from pypika import functions as fn
 
@@ -107,6 +107,8 @@ def get_stock_ledger_entries_for_batch_no(filters):
 	if not filters.get("to_date"):
 		frappe.throw(_("'To Date' is required"))
 
+	posting_datetime = get_datetime(add_to_date(filters["to_date"], days=1))
+
 	sle = frappe.qb.DocType("Stock Ledger Entry")
 	query = (
 		frappe.qb.from_(sle)
@@ -121,7 +123,7 @@ def get_stock_ledger_entries_for_batch_no(filters):
 			(sle.docstatus < 2)
 			& (sle.is_cancelled == 0)
 			& (sle.batch_no != "")
-			& (sle.posting_date <= filters["to_date"])
+			& (sle.posting_datetime < posting_datetime)
 		)
 		.groupby(sle.voucher_no, sle.batch_no, sle.item_code, sle.warehouse)
 		.orderby(sle.item_code, sle.warehouse)
diff --git a/erpnext/stock/serial_batch_bundle.py b/erpnext/stock/serial_batch_bundle.py
index 50a7707d4a9..fcebf0491ac 100644
--- a/erpnext/stock/serial_batch_bundle.py
+++ b/erpnext/stock/serial_batch_bundle.py
@@ -55,8 +55,45 @@ class SerialBatchBundle:
 		elif not self.sle.is_cancelled:
 			self.validate_item_and_warehouse()
 
+	def is_material_transfer(self):
+		allowed_types = [
+			"Material Transfer",
+			"Send to Subcontractor",
+			"Material Transfer for Manufacture",
+		]
+
+		if (
+			self.sle.voucher_type == "Stock Entry"
+			and not self.sle.is_cancelled
+			and frappe.get_cached_value("Stock Entry", self.sle.voucher_no, "purpose") in allowed_types
+		):
+			return True
+
+	def make_serial_batch_no_bundle_for_material_transfer(self):
+		from erpnext.controllers.stock_controller import make_bundle_for_material_transfer
+
+		bundle = frappe.db.get_value(
+			"Stock Entry Detail", self.sle.voucher_detail_no, "serial_and_batch_bundle"
+		)
+
+		if bundle:
+			new_bundle_id = make_bundle_for_material_transfer(
+				is_new=False,
+				docstatus=1,
+				voucher_type=self.sle.voucher_type,
+				voucher_no=self.sle.voucher_no,
+				serial_and_batch_bundle=bundle,
+				warehouse=self.sle.warehouse,
+				type_of_transaction="Inward" if self.sle.actual_qty > 0 else "Outward",
+				do_not_submit=0,
+			)
+			self.sle.db_set({"serial_and_batch_bundle": new_bundle_id})
+
 	def make_serial_batch_no_bundle(self):
 		self.validate_item()
+		if self.sle.actual_qty > 0 and self.is_material_transfer():
+			self.make_serial_batch_no_bundle_for_material_transfer()
+			return
 
 		sn_doc = SerialBatchCreation(
 			{
@@ -143,6 +180,9 @@ class SerialBatchBundle:
 				"serial_and_batch_bundle": sn_doc.name,
 			}
 
+			if self.sle.actual_qty < 0 and self.is_material_transfer():
+				values_to_update["valuation_rate"] = sn_doc.avg_rate
+
 			if not frappe.db.get_single_value(
 				"Stock Settings", "do_not_update_serial_batch_on_creation_of_auto_bundle"
 			):
@@ -341,11 +381,9 @@ def get_serial_nos(serial_and_batch_bundle, serial_nos=None):
 	if serial_nos:
 		filters["serial_no"] = ("in", serial_nos)
 
-	entries = frappe.get_all("Serial and Batch Entry", fields=["serial_no"], filters=filters, order_by="idx")
-	if not entries:
-		return []
+	serial_nos = frappe.get_all("Serial and Batch Entry", filters=filters, order_by="idx", pluck="serial_no")
 
-	return [d.serial_no for d in entries if d.serial_no]
+	return serial_nos
 
 
 def get_batches_from_bundle(serial_and_batch_bundle, batches=None):
diff --git a/erpnext/stock/stock_ledger.py b/erpnext/stock/stock_ledger.py
index ac513f2d380..7bce0dacfa5 100644
--- a/erpnext/stock/stock_ledger.py
+++ b/erpnext/stock/stock_ledger.py
@@ -312,7 +312,11 @@ def get_reposting_data(file_path) -> dict:
 	if isinstance(content, str):
 		content = content.encode("utf-8")
 
-	data = gzip.decompress(content)
+	try:
+		data = gzip.decompress(content)
+	except Exception:
+		return frappe._dict()
+
 	if data := json.loads(data.decode("utf-8")):
 		data = data
 
diff --git a/erpnext/subcontracting/doctype/subcontracting_receipt/subcontracting_receipt.js b/erpnext/subcontracting/doctype/subcontracting_receipt/subcontracting_receipt.js
index 0dff297e45d..8dfd9bd486d 100644
--- a/erpnext/subcontracting/doctype/subcontracting_receipt/subcontracting_receipt.js
+++ b/erpnext/subcontracting/doctype/subcontracting_receipt/subcontracting_receipt.js
@@ -302,6 +302,21 @@ frappe.ui.form.on("Subcontracting Receipt", {
 			};
 		}
 	},
+
+	reset_raw_materials_table: (frm) => {
+		frm.clear_table("supplied_items");
+
+		frm.call({
+			method: "reset_raw_materials",
+			doc: frm.doc,
+			freeze: true,
+			callback: (r) => {
+				if (!r.exc) {
+					frm.save();
+				}
+			},
+		});
+	},
 });
 
 frappe.ui.form.on("Landed Cost Taxes and Charges", {
diff --git a/erpnext/subcontracting/doctype/subcontracting_receipt/subcontracting_receipt.json b/erpnext/subcontracting/doctype/subcontracting_receipt/subcontracting_receipt.json
index 2024f6e5952..0013fe63219 100644
--- a/erpnext/subcontracting/doctype/subcontracting_receipt/subcontracting_receipt.json
+++ b/erpnext/subcontracting/doctype/subcontracting_receipt/subcontracting_receipt.json
@@ -47,8 +47,11 @@
   "total_qty",
   "column_break_27",
   "total",
-  "raw_material_details",
+  "raw_materials_consumed_section",
+  "reset_raw_materials_table",
+  "column_break_uinr",
   "get_current_stock",
+  "raw_material_details",
   "supplied_items",
   "additional_costs_section",
   "distribute_additional_costs_based_on",
@@ -300,6 +303,7 @@
    "depends_on": "supplied_items",
    "fieldname": "raw_material_details",
    "fieldtype": "Section Break",
+   "hide_border": 1,
    "label": "Raw Materials Consumed",
    "options": "fa fa-table",
    "print_hide": 1,
@@ -640,12 +644,26 @@
    "fieldname": "supplier_delivery_note",
    "fieldtype": "Data",
    "label": "Supplier Delivery Note"
+  },
+  {
+   "fieldname": "raw_materials_consumed_section",
+   "fieldtype": "Section Break",
+   "label": "Raw Materials Actions"
+  },
+  {
+   "fieldname": "reset_raw_materials_table",
+   "fieldtype": "Button",
+   "label": "Reset Raw Materials Table"
+  },
+  {
+   "fieldname": "column_break_uinr",
+   "fieldtype": "Column Break"
   }
  ],
  "in_create": 1,
  "is_submittable": 1,
  "links": [],
- "modified": "2024-03-27 13:10:46.856728",
+ "modified": "2024-05-27 15:02:13.517969",
  "modified_by": "Administrator",
  "module": "Subcontracting",
  "name": "Subcontracting Receipt",
diff --git a/erpnext/subcontracting/doctype/subcontracting_receipt/subcontracting_receipt.py b/erpnext/subcontracting/doctype/subcontracting_receipt/subcontracting_receipt.py
index 52193c56afc..cb0eca1b75e 100644
--- a/erpnext/subcontracting/doctype/subcontracting_receipt/subcontracting_receipt.py
+++ b/erpnext/subcontracting/doctype/subcontracting_receipt/subcontracting_receipt.py
@@ -179,6 +179,11 @@ class SubcontractingReceipt(SubcontractingController):
 		self.update_status()
 		self.delete_auto_created_batches()
 
+	@frappe.whitelist()
+	def reset_raw_materials(self):
+		self.supplied_items = []
+		self.create_raw_materials_supplied()
+
 	def validate_closed_subcontracting_order(self):
 		for item in self.items:
 			if item.subcontracting_order: