diff --git a/src/data_store.rs b/src/data_store.rs
index b1ed816df..a440a2e1e 100644
--- a/src/data_store.rs
+++ b/src/data_store.rs
@@ -81,6 +81,8 @@ where
 		Ok(updated)
 	}
 
+	/// Like [`Self::insert`], but when an entry with the object's id already exists, merges the
+	/// object's full update ([`StorableObject::to_update`]) into it instead of replacing it.
 	pub(crate) async fn insert_or_update(&self, object: SO) -> Result<bool, Error> {
 		let _guard = self.mutation_lock.lock().await;
 
@@ -170,6 +172,36 @@ where
 		Ok(DataStoreUpdateResult::Updated)
 	}
 
+	/// Atomically transforms the entry for `id` through `f` and persists the result.
+	///
+	/// `f` receives the current entry (`None` when absent) and returns the new state to write;
+	/// returning `None` leaves the store untouched. The read, the closure, and the write share
+	/// one critical section of the mutation lock, so no concurrent writer can land in between —
+	/// unlike a separate [`Self::get`] followed by an insert or update.
+	///
+	/// The closure runs on a clone of the entry with the in-memory map lock released, so it may
+	/// freely read this store or others (reads see the pre-mutation state) without ordering map
+	/// locks against each other. Keep it cheap and non-blocking.
+	///
+	/// Returns the written object, or `None` when the closure declined to write.
+	pub(crate) async fn mutate<F: FnOnce(Option<&SO>) -> Option<SO>>(
+		&self, id: &SO::Id, f: F,
+	) -> Result<Option<SO>, Error> {
+		let _guard = self.mutation_lock.lock().await;
+
+		let current = self.objects.lock().expect("lock").get(id).cloned();
+		let new_object = match f(current.as_ref()) {
+			Some(new_object) => new_object,
+			None => return Ok(None),
+		};
+		debug_assert!(new_object.id() == *id, "mutate closure must not change the object's id");
+
+		self.persist(&new_object).await?;
+		let mut locked_objects = self.objects.lock().expect("lock");
+		locked_objects.insert(new_object.id(), new_object.clone());
+		Ok(Some(new_object))
+	}
+
 	/// Returns in-memory objects matching `f`.
 	///
 	/// The async mutation lock serializes writers, but this synchronous reader cannot wait on it.
@@ -403,6 +435,131 @@ mod tests {
 		assert_eq!(Ok(true), data_store.insert_or_update(new_iou_object).await);
 	}
 
+	#[tokio::test]
+	async fn mutate_inserts_when_absent() {
+		let store: Arc<DynStore> = Arc::new(DynStoreWrapper(InMemoryStore::new()));
+		let logger = Arc::new(TestLogger::new());
+		let primary_namespace = "datastore_test_primary".to_string();
+		let secondary_namespace = "datastore_test_secondary".to_string();
+		let data_store: DataStore<TestObject, Arc<TestLogger>> = DataStore::new(
+			Vec::new(),
+			primary_namespace.clone(),
+			secondary_namespace.clone(),
+			Arc::clone(&store),
+			logger,
+		);
+
+		let id = TestObjectId { id: [42u8; 4] };
+		let object = TestObject { id, data: [23u8; 3] };
+		let result = data_store
+			.mutate(&id, |existing| {
+				assert!(existing.is_none());
+				Some(object)
+			})
+			.await;
+		assert_eq!(Ok(Some(object)), result);
+
+		assert_eq!(Some(object), data_store.get(&id));
+		let store_key = id.encode_to_hex_str();
+		assert!(KVStore::read(&*store, &primary_namespace, &secondary_namespace, &store_key)
+			.await
+			.is_ok());
+	}
+
+	#[tokio::test]
+	async fn mutate_transforms_existing_entry() {
+		let store: Arc<DynStore> = Arc::new(DynStoreWrapper(InMemoryStore::new()));
+		let logger = Arc::new(TestLogger::new());
+		let id = TestObjectId { id: [42u8; 4] };
+		let existing_object = TestObject { id, data: [23u8; 3] };
+		let data_store: DataStore<TestObject, Arc<TestLogger>> = DataStore::new(
+			vec![existing_object],
+			"datastore_test_primary".to_string(),
+			"datastore_test_secondary".to_string(),
+			store,
+			logger,
+		);
+
+		// The closure sees the current entry and derives the new state from it.
+		let result = data_store
+			.mutate(&id, |existing| {
+				let mut new_object = *existing.unwrap();
+				new_object.data[0] += 1;
+				Some(new_object)
+			})
+			.await;
+		let expected = TestObject { id, data: [24u8, 23u8, 23u8] };
+		assert_eq!(Ok(Some(expected)), result);
+		assert_eq!(Some(expected), data_store.get(&id));
+	}
+
+	#[tokio::test]
+	async fn mutate_runs_the_closure_without_the_map_lock() {
+		let store: Arc<DynStore> = Arc::new(DynStoreWrapper(InMemoryStore::new()));
+		let logger = Arc::new(TestLogger::new());
+		let id = TestObjectId { id: [42u8; 4] };
+		let existing_object = TestObject { id, data: [23u8; 3] };
+		let data_store: DataStore<TestObject, Arc<TestLogger>> = DataStore::new(
+			vec![existing_object],
+			"datastore_test_primary".to_string(),
+			"datastore_test_secondary".to_string(),
+			store,
+			logger,
+		);
+
+		// Closures gate cross-store decisions on reads of other stores, which lock their own
+		// in-memory maps. Holding this store's map lock across the closure would order it
+		// before theirs and invite lock-order inversions, so the closure must run with the map
+		// lock released.
+		let result = data_store
+			.mutate(&id, |existing| {
+				assert_eq!(Some(&existing_object), existing);
+				assert!(data_store.objects.try_lock().is_ok());
+				None
+			})
+			.await;
+		assert_eq!(Ok(None), result);
+	}
+
+	#[tokio::test]
+	async fn mutate_persists_nothing_when_closure_declines() {
+		let id = TestObjectId { id: [42u8; 4] };
+		let existing_object = TestObject { id, data: [23u8; 3] };
+		let data_store = new_failing_data_store(vec![existing_object]);
+
+		// Returning `None` must not attempt a write (the store fails all writes) nor touch memory.
+		let result = data_store
+			.mutate(&id, |existing| {
+				assert_eq!(Some(&existing_object), existing);
+				None
+			})
+			.await;
+		assert_eq!(Ok(None), result);
+		assert_eq!(Some(existing_object), data_store.get(&id));
+	}
+
+	#[tokio::test]
+	async fn mutate_does_not_mutate_memory_if_persist_fails() {
+		let existing_id = TestObjectId { id: [42u8; 4] };
+		let existing_object = TestObject { id: existing_id, data: [23u8; 3] };
+		let data_store = new_failing_data_store(vec![existing_object]);
+
+		let changed = TestObject { id: existing_id, data: [24u8; 3] };
+		assert_eq!(
+			Err(Error::PersistenceFailed),
+			data_store.mutate(&existing_id, |_| Some(changed)).await
+		);
+		assert_eq!(Some(existing_object), data_store.get(&existing_id));
+
+		let new_id = TestObjectId { id: [55u8; 4] };
+		let new_object = TestObject { id: new_id, data: [34u8; 3] };
+		assert_eq!(
+			Err(Error::PersistenceFailed),
+			data_store.mutate(&new_id, |_| Some(new_object)).await
+		);
+		assert!(data_store.get(&new_id).is_none());
+	}
+
 	#[tokio::test]
 	async fn insert_or_update_does_not_mutate_memory_if_persist_fails() {
 		let existing_id = TestObjectId { id: [42u8; 4] };
diff --git a/src/payment/pending_payment_store.rs b/src/payment/pending_payment_store.rs
index f5f2fa40a..30a113537 100644
--- a/src/payment/pending_payment_store.rs
+++ b/src/payment/pending_payment_store.rs
@@ -242,4 +242,79 @@ mod tests {
 			"current txid must not remain in its own conflict list"
 		);
 	}
+
+	#[test]
+	fn funding_classification_pending_update_preserves_mirrored_confirmation() {
+		use bitcoin::BlockHash;
+
+		use crate::payment::store::PaymentDetailsUpdate;
+
+		let txid = test_txid(7);
+		let payment_id = PaymentId(txid.to_byte_array());
+
+		// A pending entry wallet sync has already mirrored a confirmation into (via
+		// `apply_funding_status_update_locked`) before classification ran.
+		let confirmed_details = PaymentDetails::new(
+			payment_id,
+			PaymentKind::Onchain {
+				txid,
+				status: ConfirmationStatus::Confirmed {
+					block_hash: BlockHash::from_byte_array([8u8; 32]),
+					height: 100,
+					timestamp: 1,
+				},
+				tx_type: None,
+			},
+			Some(2_000_000),
+			Some(999),
+			PaymentDirection::Outbound,
+			PaymentStatus::Pending,
+		);
+		let mirrored = PendingPaymentDetails::new(confirmed_details, Vec::new(), Vec::new());
+
+		// A fresh classification is always Unconfirmed and carries the candidate history; its
+		// figures are the active candidate's.
+		let fresh = pending_onchain_payment(payment_id, txid);
+		let candidates = vec![FundingTxCandidate {
+			txid,
+			amount_msat: fresh.amount_msat,
+			fee_paid_msat: fresh.fee_paid_msat,
+		}];
+
+		// The old fresh-insert path merged the full fresh record, downgrading the mirrored
+		// confirmation.
+		let mut downgraded = mirrored.clone();
+		let full_update =
+			PendingPaymentDetails::new(fresh.clone(), Vec::new(), candidates.clone()).to_update();
+		assert!(downgraded.update(full_update));
+		assert!(
+			matches!(
+				downgraded.details.kind,
+				PaymentKind::Onchain { status: ConfirmationStatus::Unconfirmed, .. }
+			),
+			"a full merge of a fresh classification downgrades a mirrored confirmation",
+		);
+
+		// The narrow classification update merges the candidates while preserving the
+		// confirmation state wallet sync owns. It names the confirmed txid, so its
+		// contribution-derived figures replace the mirrored wallet-view ones.
+		let mut merged = mirrored.clone();
+		let narrow_update = PendingPaymentDetailsUpdate {
+			id: payment_id,
+			payment_update: Some(PaymentDetailsUpdate::funding_reclassification(fresh)),
+			conflicting_txids: None,
+			candidates: candidates.clone(),
+		};
+		assert!(merged.update(narrow_update));
+		assert!(
+			matches!(
+				merged.details.kind,
+				PaymentKind::Onchain { status: ConfirmationStatus::Confirmed { .. }, .. }
+			),
+			"a narrow classification update must not downgrade a mirrored confirmation",
+		);
+		assert_eq!(merged.candidates, candidates);
+		assert_eq!(merged.details.amount_msat, Some(1_000));
+		assert_eq!(merged.details.fee_paid_msat, Some(100));
+	}
 }
diff --git a/src/payment/store.rs b/src/payment/store.rs
index d2b92747a..e0f3f93a8 100644
--- a/src/payment/store.rs
+++ b/src/payment/store.rs
@@ -243,12 +243,28 @@ impl StorableObject for PaymentDetails {
 			}
 		}
 
-		if let Some(amount_opt) = update.amount_msat {
-			update_if_necessary!(self.amount_msat, amount_opt);
-		}
+		// Once an on-chain record is confirmed, its txid and figures describe the candidate that
+		// confirmed, which need not be the last one broadcast. An update that doesn't assert the
+		// confirmation state was built without knowing it — e.g. a late funding classification
+		// whose candidate lost to the counterparty's broadcast — so it must not move them. The
+		// exception is an update naming the confirmed txid itself: its figures describe the very
+		// candidate that confirmed and correct the wallet-view amount/fee a sync-created record
+		// carries, which cannot represent our contribution to a shared funding output.
+		let keep_confirmed_figures = update.confirmation_status.is_none()
+			&& matches!(
+				self.kind,
+				PaymentKind::Onchain { txid, status: ConfirmationStatus::Confirmed { .. }, .. }
+					if update.txid != Some(txid)
+			);
+
+		if !keep_confirmed_figures {
+			if let Some(amount_opt) = update.amount_msat {
+				update_if_necessary!(self.amount_msat, amount_opt);
+			}
 
-		if let Some(fee_paid_msat_opt) = update.fee_paid_msat {
-			update_if_necessary!(self.fee_paid_msat, fee_paid_msat_opt);
+			if let Some(fee_paid_msat_opt) = update.fee_paid_msat {
+				update_if_necessary!(self.fee_paid_msat, fee_paid_msat_opt);
+			}
 		}
 
 		if let Some(skimmed_fee_msat) = update.counterparty_skimmed_fee_msat {
@@ -278,7 +294,7 @@ impl StorableObject for PaymentDetails {
 
 		if let Some(tx_id) = update.txid {
 			match self.kind {
-				PaymentKind::Onchain { ref mut txid, .. } => {
+				PaymentKind::Onchain { ref mut txid, .. } if !keep_confirmed_figures => {
 					update_if_necessary!(*txid, tx_id);
 				},
 				_ => {},
@@ -712,6 +728,33 @@ impl PaymentDetailsUpdate {
 			tx_type: None,
 		}
 	}
+
+	/// Builds an update that merges a freshly-classified funding payment's classification
+	/// (`tx_type`), broadcast txid, and our contribution figures (amount/fee) into an existing
+	/// record, while leaving the top-level [`PaymentStatus`] and the on-chain
+	/// [`ConfirmationStatus`] untouched.
+	///
+	/// Funding classification runs off the broadcaster queue and can land *after* wallet sync has
+	/// already advanced a record's confirmation state (e.g. when the counterparty's broadcast of
+	/// the funding transaction is observed first). Merging only the funding-specific fields keeps
+	/// such a late classification from downgrading a `Confirmed`/`Succeeded` payment back to
+	/// `Unconfirmed`/`Pending`; the confirmation state is owned by the wallet-sync events instead.
+	///
+	/// The txid and figures are taken from the freshly broadcast (active) candidate, so they only
+	/// apply while the record is unconfirmed. Once a candidate confirms, the record's txid and
+	/// figures describe that candidate — which need not be the one being classified (e.g. the
+	/// counterparty broadcast an earlier candidate and it won) — and [`PaymentDetails::update`]
+	/// leaves them in place for updates like this one that don't carry a confirmation state.
+	pub(crate) fn funding_reclassification(details: PaymentDetails) -> Self {
+		let mut update = Self::new(details.id);
+		update.amount_msat = Some(details.amount_msat);
+		update.fee_paid_msat = Some(details.fee_paid_msat);
+		if let PaymentKind::Onchain { txid, tx_type, .. } = details.kind {
+			update.txid = Some(txid);
+			update.tx_type = Some(tx_type);
+		}
+		update
+	}
 }
 
 impl From<&PaymentDetails> for PaymentDetailsUpdate {
@@ -1022,6 +1065,370 @@ mod tests {
 		}
 	}
 
+	#[test]
+	fn funding_reclassification_does_not_downgrade_an_advanced_record() {
+		use bitcoin::hashes::Hash;
+		use std::str::FromStr;
+
+		// A splice funding payment wallet sync has already advanced to Succeeded/Confirmed.
+		let txid = Txid::from_byte_array([7u8; 32]);
+		let id = PaymentId(txid.to_byte_array());
+		let tx_type = Some(TransactionType::InteractiveFunding {
+			channels: vec![Channel {
+				counterparty_node_id: PublicKey::from_str(
+					"0279be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798",
+				)
+				.unwrap(),
+				channel_id: ChannelId([3u8; 32]),
+			}],
+		});
+		let advanced = PaymentDetails::new(
+			id,
+			PaymentKind::Onchain {
+				txid,
+				status: ConfirmationStatus::Confirmed {
+					block_hash: BlockHash::from_byte_array([8u8; 32]),
+					height: 100,
+					timestamp: 1,
+				},
+				tx_type: tx_type.clone(),
+			},
+			Some(2_000_000),
+			Some(999),
+			PaymentDirection::Outbound,
+			PaymentStatus::Succeeded,
+		);
+
+		// A fresh funding classification for the same payment is always Pending/Unconfirmed.
+		let fresh = PaymentDetails::new(
+			id,
+			PaymentKind::Onchain { txid, status: ConfirmationStatus::Unconfirmed, tx_type },
+			Some(1_000_000),
+			Some(500),
+			PaymentDirection::Outbound,
+			PaymentStatus::Pending,
+		);
+
+		// The naive full update `insert_or_update` applied before the fix downgrades both the
+		// top-level status and the on-chain confirmation status — the bug Codex flagged.
+		let mut downgraded = advanced.clone();
+		downgraded.update((&fresh).into());
+		assert_eq!(
+			downgraded.status,
+			PaymentStatus::Pending,
+			"a full update from a fresh classification downgrades the top-level status",
+		);
+		assert!(
+			matches!(
+				downgraded.kind,
+				PaymentKind::Onchain { status: ConfirmationStatus::Unconfirmed, .. }
+			),
+			"a full update from a fresh classification downgrades the confirmation status",
+		);
+
+		// The narrowed reclassification update merges only the funding fields and preserves the
+		// advanced confirmation state that wallet sync owns.
+		let mut merged = advanced.clone();
+		merged.update(PaymentDetailsUpdate::funding_reclassification(fresh));
+		assert_eq!(
+			merged.status,
+			PaymentStatus::Succeeded,
+			"reclassification must not downgrade the top-level status",
+		);
+		assert!(
+			matches!(
+				merged.kind,
+				PaymentKind::Onchain {
+					status: ConfirmationStatus::Confirmed { .. },
+					tx_type: Some(TransactionType::InteractiveFunding { .. }),
+					..
+				}
+			),
+			"reclassification must preserve the confirmation status and keep the funding tx_type",
+		);
+		// The late classification names the confirmed txid, so its contribution-derived figures
+		// replace the record's; only an update for a different candidate leaves them in place
+		// (covered by `funding_reclassification_keeps_confirmed_candidate_figures`).
+		assert_eq!(merged.amount_msat, Some(1_000_000));
+		assert_eq!(merged.fee_paid_msat, Some(500));
+	}
+
+	#[test]
+	fn funding_reclassification_keeps_confirmed_candidate_figures() {
+		use bitcoin::hashes::Hash;
+		use std::str::FromStr;
+
+		// A funding payment whose first candidate wallet sync has already seen confirm — e.g. the
+		// counterparty's broadcast of it was picked up before our own later candidate was
+		// classified. The record is unclassified (created by the sync fallthrough).
+		let confirmed_txid = Txid::from_byte_array([7u8; 32]);
+		let id = PaymentId(confirmed_txid.to_byte_array());
+		let confirmed = PaymentDetails::new(
+			id,
+			PaymentKind::Onchain {
+				txid: confirmed_txid,
+				status: ConfirmationStatus::Confirmed {
+					block_hash: BlockHash::from_byte_array([8u8; 32]),
+					height: 100,
+					timestamp: 1,
+				},
+				tx_type: None,
+			},
+			Some(2_000_000),
+			Some(999),
+			PaymentDirection::Outbound,
+			PaymentStatus::Pending,
+		);
+
+		// Our own, different (e.g. fee-bumped) candidate is classified late.
+		let late_txid = Txid::from_byte_array([9u8; 32]);
+		let tx_type = Some(TransactionType::InteractiveFunding {
+			channels: vec![Channel {
+				counterparty_node_id: PublicKey::from_str(
+					"0279be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798",
+				)
+				.unwrap(),
+				channel_id: ChannelId([3u8; 32]),
+			}],
+		});
+		let late = PaymentDetails::new(
+			id,
+			PaymentKind::Onchain {
+				txid: late_txid,
+				status: ConfirmationStatus::Unconfirmed,
+				tx_type,
+			},
+			Some(1_000_000),
+			Some(500),
+			PaymentDirection::Outbound,
+			PaymentStatus::Pending,
+		);
+
+		// The confirmed record's txid and figures describe the candidate that confirmed; the late
+		// classification must not replace them with an unconfirmed candidate's. The
+		// classification itself (`tx_type`) still lands.
+		let mut classified = confirmed.clone();
+		classified.update(PaymentDetailsUpdate::funding_reclassification(late.clone()));
+		assert!(
+			matches!(
+				classified.kind,
+				PaymentKind::Onchain {
+					txid,
+					tx_type: Some(TransactionType::InteractiveFunding { .. }),
+					..
+				} if txid == confirmed_txid
+			),
+			"a late classification must set the tx_type but not replace a confirmed record's txid",
+		);
+		assert_eq!(classified.amount_msat, Some(2_000_000));
+		assert_eq!(classified.fee_paid_msat, Some(999));
+
+		// While the record is still unconfirmed, the freshly broadcast candidate is the active
+		// one, so its txid and figures do replace the stored ones (RBF rotation).
+		let mut unconfirmed = confirmed.clone();
+		if let PaymentKind::Onchain { ref mut status, .. } = unconfirmed.kind {
+			*status = ConfirmationStatus::Unconfirmed;
+		}
+		unconfirmed.update(PaymentDetailsUpdate::funding_reclassification(late));
+		assert!(
+			matches!(unconfirmed.kind, PaymentKind::Onchain { txid, .. } if txid == late_txid),
+			"classifying a new candidate of an unconfirmed record rotates the txid",
+		);
+		assert_eq!(unconfirmed.amount_msat, Some(1_000_000));
+		assert_eq!(unconfirmed.fee_paid_msat, Some(500));
+	}
+
+	#[test]
+	fn funding_reclassification_merges_figures_for_the_confirmed_candidate() {
+		use bitcoin::hashes::Hash;
+		use std::str::FromStr;
+
+		// Wallet sync confirmed the transaction before classification ran, so the record carries
+		// the wallet's own view of amount/fee, which cannot represent our contribution to a shared
+		// funding output.
+		let confirmed_txid = Txid::from_byte_array([7u8; 32]);
+		let id = PaymentId(confirmed_txid.to_byte_array());
+		let mut record = PaymentDetails::new(
+			id,
+			PaymentKind::Onchain {
+				txid: confirmed_txid,
+				status: ConfirmationStatus::Confirmed {
+					block_hash: BlockHash::from_byte_array([8u8; 32]),
+					height: 100,
+					timestamp: 1,
+				},
+				tx_type: None,
+			},
+			Some(2_000_000),
+			Some(999),
+			PaymentDirection::Outbound,
+			PaymentStatus::Pending,
+		);
+
+		// The late classification names the candidate that confirmed, so its contribution-derived
+		// figures are authoritative and must replace the wallet-view ones; only an update for a
+		// different (losing) candidate leaves a confirmed record's figures in place.
+		let tx_type = Some(TransactionType::InteractiveFunding {
+			channels: vec![Channel {
+				counterparty_node_id: PublicKey::from_str(
+					"0279be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798",
+				)
+				.unwrap(),
+				channel_id: ChannelId([3u8; 32]),
+			}],
+		});
+		let classified = PaymentDetails::new(
+			id,
+			PaymentKind::Onchain {
+				txid: confirmed_txid,
+				status: ConfirmationStatus::Unconfirmed,
+				tx_type,
+			},
+			Some(1_000_000),
+			Some(500),
+			PaymentDirection::Outbound,
+			PaymentStatus::Pending,
+		);
+
+		assert!(record.update(PaymentDetailsUpdate::funding_reclassification(classified)));
+		assert!(
+			matches!(
+				record.kind,
+				PaymentKind::Onchain {
+					txid,
+					status: ConfirmationStatus::Confirmed { .. },
+					tx_type: Some(TransactionType::InteractiveFunding { .. }),
+				} if txid == confirmed_txid
+			),
+			"the confirmed txid, confirmation state, and classification must all be in place",
+		);
+		assert_eq!(record.amount_msat, Some(1_000_000));
+		assert_eq!(record.fee_paid_msat, Some(500));
+	}
+
+	#[tokio::test]
+	async fn funding_classification_merge_preserves_advanced_record() {
+		use bitcoin::hashes::Hash;
+		use lightning::util::test_utils::TestLogger;
+		use std::str::FromStr;
+		use std::sync::Arc;
+
+		use crate::data_store::DataStore;
+		use crate::io::test_utils::InMemoryStore;
+		use crate::types::{DynStore, DynStoreWrapper};
+
+		let txid = Txid::from_byte_array([7u8; 32]);
+		let id = PaymentId(txid.to_byte_array());
+		let tx_type = Some(TransactionType::InteractiveFunding {
+			channels: vec![Channel {
+				counterparty_node_id: PublicKey::from_str(
+					"0279be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798",
+				)
+				.unwrap(),
+				channel_id: ChannelId([3u8; 32]),
+			}],
+		});
+		// A funding payment wallet sync has already recorded (unclassified, via the default
+		// on-chain path) and advanced to Succeeded/Confirmed.
+		let advanced = PaymentDetails::new(
+			id,
+			PaymentKind::Onchain {
+				txid,
+				status: ConfirmationStatus::Confirmed {
+					block_hash: BlockHash::from_byte_array([8u8; 32]),
+					height: 100,
+					timestamp: 1,
+				},
+				tx_type: None,
+			},
+			Some(2_000_000),
+			Some(999),
+			PaymentDirection::Outbound,
+			PaymentStatus::Succeeded,
+		);
+		// A fresh funding classification for the same payment is always Pending/Unconfirmed.
+		let fresh = PaymentDetails::new(
+			id,
+			PaymentKind::Onchain { txid, status: ConfirmationStatus::Unconfirmed, tx_type },
+			Some(1_000_000),
+			Some(500),
+			PaymentDirection::Outbound,
+			PaymentStatus::Pending,
+		);
+
+		let new_store = |seed: Vec<PaymentDetails>| {
+			let store: Arc<DynStore> = Arc::new(DynStoreWrapper(InMemoryStore::new()));
+			let logger = Arc::new(TestLogger::new());
+			DataStore::<PaymentDetails, Arc<TestLogger>>::new(
+				seed,
+				"payment_test_primary".to_string(),
+				"payment_test_secondary".to_string(),
+				store,
+				logger,
+			)
+		};
+
+		// The pre-fix fresh-insert path — a full `insert_or_update` merge landing after a racing
+		// wallet sync already advanced the record — downgrades it.
+		let store = new_store(vec![advanced.clone()]);
+		store.insert_or_update(fresh.clone()).await.unwrap();
+		let downgraded = store.get(&id).unwrap();
+		assert_eq!(
+			downgraded.status,
+			PaymentStatus::Pending,
+			"a full merge of a fresh classification downgrades an advanced record",
+		);
+
+		// Classification instead applies only the narrow reclassification when a record exists —
+		// no matter when it appeared — setting the `tx_type` while preserving the confirmation
+		// state wallet sync owns. The update names the confirmed txid, so its
+		// contribution-derived figures replace the record's wallet-view ones.
+		let store = new_store(vec![advanced.clone()]);
+		let update = PaymentDetailsUpdate::funding_reclassification(fresh.clone());
+		let written = store
+			.mutate(&id, |existing| match existing {
+				Some(current) => {
+					let mut updated = current.clone();
+					updated.update(update).then_some(updated)
+				},
+				None => Some(fresh.clone()),
+			})
+			.await;
+		assert!(matches!(written, Ok(Some(_))), "the reclassification must merge");
+		let merged = store.get(&id).unwrap();
+		assert_eq!(merged.status, PaymentStatus::Succeeded);
+		assert!(matches!(
+			merged.kind,
+			PaymentKind::Onchain {
+				status: ConfirmationStatus::Confirmed { .. },
+				tx_type: Some(TransactionType::InteractiveFunding { .. }),
+				..
+			}
+		));
+		assert_eq!(merged.amount_msat, Some(1_000_000));
+		assert_eq!(merged.fee_paid_msat, Some(500));
+
+		// And it inserts the fresh details when no record exists yet.
+		let store = new_store(Vec::new());
+		let update = PaymentDetailsUpdate::funding_reclassification(fresh.clone());
+		let written = store
+			.mutate(&id, |existing| match existing {
+				Some(current) => {
+					let mut updated = current.clone();
+					updated.update(update).then_some(updated)
+				},
+				None => Some(fresh.clone()),
+			})
+			.await;
+		assert!(matches!(written, Ok(Some(_))), "the fresh details must insert");
+		let inserted = store.get(&id).unwrap();
+		assert_eq!(inserted.status, PaymentStatus::Pending);
+		assert!(matches!(
+			inserted.kind,
+			PaymentKind::Onchain { status: ConfirmationStatus::Unconfirmed, .. }
+		));
+	}
+
 	#[derive(Clone, Debug, PartialEq, Eq)]
 	struct LegacyBolt11JitKind {
 		hash: PaymentHash,
diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs
index f8d9d521e..8d9c4e6d2 100644
--- a/src/wallet/mod.rs
+++ b/src/wallet/mod.rs
@@ -54,9 +54,11 @@ use lightning_invoice::RawBolt11Invoice;
 use persist::KVStoreWalletPersister;
 
 use crate::config::Config;
+use crate::data_store::StorableObject;
 use crate::fee_estimator::{ConfirmationTarget, FeeEstimator, OnchainFeeEstimator};
 use crate::logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger};
-use crate::payment::store::ConfirmationStatus;
+use crate::payment::pending_payment_store::PendingPaymentDetailsUpdate;
+use crate::payment::store::{ConfirmationStatus, PaymentDetailsUpdate};
 use crate::payment::{
 	FundingTxCandidate, PaymentDetails, PaymentDirection, PaymentKind, PaymentStatus,
 	PendingPaymentDetails, TransactionType,
@@ -93,6 +95,17 @@ pub(crate) struct Wallet {
 	config: Arc<Config>,
 	logger: Arc<Logger>,
 	pending_payment_store: Arc<PendingPaymentStore>,
+	// Serializes the writers that must observe the payment record and its pending-store entry
+	// (candidate history included) as one consistent unit: classification holds it across its
+	// two-store write pair, and wallet sync's event arms hold it from payment-id resolution
+	// through their last write. Without it, a confirmation landing between classification's two
+	// writes sees the record classified but the candidate history absent — resolving the wrong
+	// payment id or stamping the confirmed candidate with another candidate's figures — and a
+	// classification landing inside an arm's decision sequence gets overwritten by the arm's
+	// stale generic fallback. Graduation stays off this lock: it decides from the live record
+	// under the payment store's mutation lock and writes only the status, so it carries nothing
+	// a concurrent classification could lose.
+	funding_payment_update_lock: tokio::sync::Mutex<()>,
 }
 
 impl Wallet {
@@ -116,6 +129,7 @@ impl Wallet {
 			config,
 			logger,
 			pending_payment_store,
+			funding_payment_update_lock: tokio::sync::Mutex::new(()),
 		}
 	}
 
@@ -255,12 +269,23 @@ impl Wallet {
 						timestamp: block_time.confirmation_time,
 					};
 
+					// Hold the cross-store lock from payment-id resolution through the last write:
+					// a classification landing in between would leave the id resolved against a
+					// torn candidate index and the generic fallback below overwriting (or
+					// duplicating) the record classification just wrote.
+					let guard = self.funding_payment_update_lock.lock().await;
+
 					let payment_id = self
 						.find_payment_by_txid(txid)
 						.unwrap_or_else(|| PaymentId(txid.to_byte_array()));
 
 					if self
-						.apply_funding_status_update(payment_id, txid, confirmation_status)
+						.apply_funding_status_update_locked(
+							&guard,
+							payment_id,
+							txid,
+							confirmation_status,
+						)
 						.await?
 					{
 						continue;
@@ -301,7 +326,7 @@ impl Wallet {
 
 					let mut unconfirmed_outbound_txids: Vec<Txid> = Vec::new();
 
-					for mut payment in pending_payments {
+					for payment in pending_payments {
 						match payment.details.kind {
 							PaymentKind::Onchain {
 								status: ConfirmationStatus::Confirmed { height, .. },
@@ -309,9 +334,41 @@ impl Wallet {
 							} => {
 								let payment_id = payment.details.id;
 								if new_tip.height >= height + ANTI_REORG_DELAY - 1 {
-									payment.details.status = PaymentStatus::Succeeded;
-									self.payment_store.insert_or_update(payment.details).await?;
-									self.pending_payment_store.remove(&payment_id).await?;
+									// Graduate from the live record, not the snapshot listed
+									// above: a classification landing since then must not have
+									// its figures rolled back. The status-only update carries
+									// no figures/txid/confirmation, so nothing a concurrent
+									// writer wrote can be clobbered; the update machinery bumps
+									// `latest_update_timestamp` and no-ops when the record is
+									// already `Succeeded`. A record that has diverged from the
+									// snapshot (or was removed) declines, leaving future
+									// events to drive it.
+									let mut graduated = false;
+									self.payment_store
+										.mutate(&payment_id, |existing| {
+											let current = existing?;
+											match current.kind {
+												PaymentKind::Onchain {
+													status:
+														ConfirmationStatus::Confirmed { height, .. },
+													..
+												} if new_tip.height
+													>= height + ANTI_REORG_DELAY - 1 =>
+												{
+													graduated = true;
+													let mut update =
+														PaymentDetailsUpdate::new(payment_id);
+													update.status = Some(PaymentStatus::Succeeded);
+													let mut updated = current.clone();
+													updated.update(update).then_some(updated)
+												},
+												_ => None,
+											}
+										})
+										.await?;
+									if graduated {
+										self.pending_payment_store.remove(&payment_id).await?;
+									}
 								}
 							},
 							PaymentKind::Onchain {
@@ -352,12 +409,17 @@ impl Wallet {
 					}
 				},
 				WalletEvent::TxUnconfirmed { txid, tx, .. } => {
+					// See `TxConfirmed`: id resolution and the writes below must not interleave
+					// with classification.
+					let guard = self.funding_payment_update_lock.lock().await;
+
 					let payment_id = self
 						.find_payment_by_txid(txid)
 						.unwrap_or_else(|| PaymentId(txid.to_byte_array()));
 
 					if self
-						.apply_funding_status_update(
+						.apply_funding_status_update_locked(
+							&guard,
 							payment_id,
 							txid,
 							ConfirmationStatus::Unconfirmed,
@@ -384,6 +446,12 @@ impl Wallet {
 					self.pending_payment_store.insert_or_update(pending_payment).await?;
 				},
 				WalletEvent::TxReplaced { txid, conflicts, .. } => {
+					// See `TxConfirmed`: id resolution and the writes below must not interleave
+					// with classification. The pending entry written below embeds a read of the
+					// payment record, which must not go stale against a concurrent
+					// classification either.
+					let _guard = self.funding_payment_update_lock.lock().await;
+
 					let Some(payment_id) = self.find_payment_by_txid(txid) else {
 						log_error!(
 							self.logger,
@@ -398,9 +466,11 @@ impl Wallet {
 						conflicts.iter().map(|(_, conflict_txid)| *conflict_txid).collect();
 
 					conflict_txids.push(txid);
-					// The payment already exists in the store at this point: `bump_fee_rbf` updates
-					// the payment store with the replacement txid before the next sync cycle, so we
-					// can safely fetch it here.
+					// The payment already exists in the store at this point: `bump_fee_rbf`
+					// updates the payment store with the replacement txid before the next sync
+					// cycle, and an id resolved through the candidate history comes from a
+					// classification whose payment-store write strictly precedes the candidate
+					// history it was resolved from. So we can safely fetch it here.
 					debug_assert!(
 						self.payment_store.get(&payment_id).is_some(),
 						"Payment {:?} expected in store during WalletEvent::TxReplaced but not found",
@@ -414,12 +484,17 @@ impl Wallet {
 					self.pending_payment_store.insert_or_update(pending_payment_details).await?;
 				},
 				WalletEvent::TxDropped { txid, tx } => {
+					// See `TxConfirmed`: id resolution and the writes below must not interleave
+					// with classification.
+					let guard = self.funding_payment_update_lock.lock().await;
+
 					let payment_id = self
 						.find_payment_by_txid(txid)
 						.unwrap_or_else(|| PaymentId(txid.to_byte_array()));
 
 					if self
-						.apply_funding_status_update(
+						.apply_funding_status_update_locked(
+							&guard,
 							payment_id,
 							txid,
 							ConfirmationStatus::Unconfirmed,
@@ -1398,9 +1473,82 @@ impl Wallet {
 	async fn persist_funding_payment(
 		&self, details: PaymentDetails, candidates: Vec<FundingTxCandidate>,
 	) -> Result<(), Error> {
-		self.payment_store.insert_or_update(details.clone()).await?;
-		let pending = PendingPaymentDetails::new(details, Vec::new(), candidates);
-		self.pending_payment_store.insert_or_update(pending).await?;
+		// Hold the cross-store lock across both writes so a funding confirmation never observes
+		// the record classified but the candidate history it needs still missing.
+		let _guard = self.funding_payment_update_lock.lock().await;
+
+		// Everything this write does depends on the record's current state, so all of it must be
+		// decided inside the store's critical section. When a record exists — no matter when it
+		// appeared — only the classification (`tx_type`) and the figures of whichever candidate
+		// the record's state makes authoritative are merged: a full merge of the fresh
+		// Pending/Unconfirmed details would downgrade the confirmation state the wallet-sync
+		// events own. Which candidate is authoritative is equally stateful: substituting the
+		// confirmed candidate's figures requires seeing the confirmation. Selected from a read
+		// taken before the lock, the choice goes stale when a confirmation lands in between —
+		// the update still names the actively-broadcast candidate, the confirmed-figures guard
+		// then rightly refuses it, and the record is left with figures no classification derived.
+		let id = details.id;
+		let mut update = None;
+		self.payment_store
+			.mutate(&id, |existing| {
+				let reclassification =
+					funding_reclassification_update(details.clone(), &candidates, existing);
+				update = Some(reclassification.clone());
+				match existing {
+					None => Some(details.clone()),
+					Some(current) => {
+						let mut updated = current.clone();
+						updated.update(reclassification).then_some(updated)
+					},
+				}
+			})
+			.await?;
+		let update = update.expect("the mutate closure always runs");
+
+		// The pending index must exist exactly while the authoritative record is Pending:
+		// graduation and rebroadcast read it, and a graduated payment must not be re-indexed.
+		// Deciding by the post-write status rather than by whether the write inserted also
+		// repairs a missing index — a crash or failed write between the two stores leaves a
+		// Pending record with no entry, and a merge alone would never recreate it, leaving the
+		// payment unable to graduate and its txids unmapped.
+		//
+		// The status must be read inside the pending store's critical section. Graduation writes
+		// `Succeeded` before removing the entry, so a read there that still observes `Pending`
+		// is ordered before the removal, which then also deletes anything inserted here. A
+		// status read taken before this write goes stale when graduation lands in between, and
+		// would re-index the graduated payment.
+		self.pending_payment_store
+			.mutate(&id, |existing| {
+				// The record was written above and payment records are never removed, so absence
+				// means the write failed out; fall back to the fresh details.
+				let recorded = self.payment_store.get(&id).unwrap_or(details);
+				match existing {
+					// The inserted entry embeds the post-write record rather than the fresh
+					// details, so a confirmation wallet sync already recorded keeps driving
+					// graduation.
+					None if recorded.status == PaymentStatus::Pending => {
+						Some(PendingPaymentDetails::new(recorded, Vec::new(), candidates))
+					},
+					// The payment already advanced beyond Pending: the graduation path removed
+					// the entry and it must not be re-created.
+					None => None,
+					// The entry predates this classification — wallet sync recorded the
+					// transaction before it was classified (its arms and this write pair
+					// serialize on the cross-store lock, so nothing lands in between): merge
+					// only the classification into the existing entry.
+					Some(entry) => {
+						let pending_update = PendingPaymentDetailsUpdate {
+							id,
+							payment_update: Some(update),
+							conflicting_txids: None,
+							candidates,
+						};
+						let mut updated = entry.clone();
+						updated.update(pending_update).then_some(updated)
+					},
+				}
+			})
+			.await?;
 		Ok(())
 	}
 
@@ -1483,6 +1631,10 @@ impl Wallet {
 			.list_filter(|p| {
 				matches!(p.details.kind, PaymentKind::Onchain { txid, .. } if txid == target_txid)
 					|| p.conflicting_txids.contains(&target_txid)
+					// A middle RBF round is not the record's current txid and may never have
+					// received a `TxReplaced` event of its own, so map any of its candidate
+					// txids (an earlier RBF round may confirm) back to the record.
+					|| p.candidate(target_txid).is_some()
 			})
 			.first()
 		{
@@ -1498,37 +1650,63 @@ impl Wallet {
 	/// `sent`/`received` don't capture our contribution to a shared funding output. Returns `true`
 	/// when it handled the payment, so the caller skips the default on-chain path. Graduation to
 	/// `Succeeded` is left to `ChainTipChanged` after `ANTI_REORG_DELAY`.
-	async fn apply_funding_status_update(
-		&self, payment_id: PaymentId, event_txid: Txid, confirmation_status: ConfirmationStatus,
+	///
+	/// The caller must hold [`Self::funding_payment_update_lock`] — from resolving `payment_id`
+	/// through its own last write, not just across this call — so that classification's two-store
+	/// write pair cannot interleave with the caller's decision sequence. The `_guard` parameter
+	/// serves as a reminder of that contract.
+	async fn apply_funding_status_update_locked(
+		&self, _guard: &tokio::sync::MutexGuard<'_, ()>, payment_id: PaymentId, event_txid: Txid,
+		confirmation_status: ConfirmationStatus,
 	) -> Result<bool, Error> {
-		let Some(mut payment) = self.payment_store.get(&payment_id) else {
+		// The funding-type gate, the candidate lookup, and the write share the store's mutation
+		// lock: against a separate `get`, a classification merging in between would have its
+		// `tx_type` and contribution figures clobbered by this stale snapshot.
+		let mut handled = None;
+		self.payment_store
+			.mutate(&payment_id, |existing| {
+				let payment = existing?;
+				let tx_type = match &payment.kind {
+					PaymentKind::Onchain {
+						tx_type:
+							tx_type @ Some(
+								TransactionType::Funding { .. }
+								| TransactionType::InteractiveFunding { .. },
+							),
+						..
+					} => tx_type.clone(),
+					_ => return None,
+				};
+				// Report the figures of the candidate that actually confirmed, which need not be
+				// the last one broadcast (an earlier, lower-fee candidate may win) and may carry
+				// no figures at all (`None`) for a round we didn't contribute to. (`direction` is
+				// invariant across a splice's candidates and cannot be changed through the store
+				// anyway.)
+				let mut target = payment.clone();
+				if let Some(pending) = self.pending_payment_store.get(&payment_id) {
+					if let Some(candidate) = pending.candidate(event_txid) {
+						target.amount_msat = candidate.amount_msat;
+						target.fee_paid_msat = candidate.fee_paid_msat;
+					}
+				}
+				target.kind =
+					PaymentKind::Onchain { txid: event_txid, status: confirmation_status, tx_type };
+
+				// Merge through the update machinery so its rules (e.g. which fields a merge may
+				// touch) keep applying, and skip the write when nothing changed.
+				let mut merged = payment.clone();
+				if merged.update(target.to_update()) {
+					handled = Some(merged.clone());
+					Some(merged)
+				} else {
+					handled = Some(payment.clone());
+					None
+				}
+			})
+			.await?;
+		let Some(payment) = handled else {
 			return Ok(false);
 		};
-		let tx_type = match &payment.kind {
-			PaymentKind::Onchain {
-				tx_type:
-					tx_type @ Some(
-						TransactionType::Funding { .. }
-						| TransactionType::InteractiveFunding { .. },
-					),
-				..
-			} => tx_type.clone(),
-			_ => return Ok(false),
-		};
-		// Report the figures of the candidate that actually confirmed, which need not be the last
-		// one broadcast (an earlier, lower-fee candidate may win) and may carry no figures at all
-		// (`None`) for a round we didn't contribute to. (`direction` is invariant across a splice's
-		// candidates and cannot be changed through the store anyway.)
-		if let Some(pending) = self.pending_payment_store.get(&payment_id) {
-			if let Some(candidate) = pending.candidate(event_txid) {
-				payment.amount_msat = candidate.amount_msat;
-				payment.fee_paid_msat = candidate.fee_paid_msat;
-			}
-		}
-
-		payment.kind =
-			PaymentKind::Onchain { txid: event_txid, status: confirmation_status, tx_type };
-		self.payment_store.insert_or_update(payment.clone()).await?;
 		// Mirror the refreshed confirmation status onto the pending entry: `ChainTipChanged`
 		// graduates by reading the pending entry's details, so it must see the new status. This is
 		// the same dual-write the default `TxConfirmed` path performs; an empty conflicting-txids
@@ -2154,3 +2332,608 @@ fn ldk_to_bdk_satisfaction_weight(ldk_satisfaction_weight: u64) -> Weight {
 			.saturating_sub(EMPTY_SCRIPT_SIG_WEIGHT + EMPTY_WITNESS_COUNT_WEIGHT),
 	)
 }
+
+/// Builds the payment-store update for a freshly classified funding payment. `details` describes
+/// the actively broadcast candidate, but when the record already confirmed a *different*
+/// candidate — wallet sync saw it win before this classification ran — the update instead carries
+/// the confirmed candidate's txid and figures from the candidate history, mirroring what
+/// [`Wallet::apply_funding_status_update_locked`] reports when confirmation arrives after
+/// classification.
+///
+/// `current` is the record as observed inside the payment store's `mutate` critical section — its
+/// sole caller, [`Wallet::persist_funding_payment`], builds and applies the update within one
+/// closure — so the candidate choice cannot go stale against a concurrent confirmation before the
+/// update lands. [`PaymentDetails::update`]'s confirmed-figures rule still arbitrates which
+/// figures may land on the record.
+fn funding_reclassification_update(
+	details: PaymentDetails, candidates: &[FundingTxCandidate], current: Option<&PaymentDetails>,
+) -> PaymentDetailsUpdate {
+	let mut update = PaymentDetailsUpdate::funding_reclassification(details);
+	if let Some(PaymentKind::Onchain {
+		txid: confirmed_txid,
+		status: ConfirmationStatus::Confirmed { .. },
+		..
+	}) = current.map(|payment| &payment.kind)
+	{
+		if update.txid != Some(*confirmed_txid) {
+			if let Some(candidate) = candidates.iter().find(|c| c.txid == *confirmed_txid) {
+				update.txid = Some(candidate.txid);
+				update.amount_msat = Some(candidate.amount_msat);
+				update.fee_paid_msat = Some(candidate.fee_paid_msat);
+			}
+		}
+	}
+	update
+}
+
+#[cfg(test)]
+mod tests {
+	use std::time::Duration;
+
+	use bdk_chain::{BlockId, ConfirmationBlockTime};
+	use bitcoin::hashes::Hash;
+	use bitcoin::Network;
+	use lightning::io;
+	use lightning::util::persist::{KVStore, PageToken, PaginatedKVStore, PaginatedListResponse};
+
+	use super::*;
+	use crate::config::EsploraSyncConfig;
+	use crate::io::test_utils::InMemoryStore;
+	use crate::io::{
+		PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
+		PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
+		PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
+	};
+	use crate::tx_broadcaster::TransactionBroadcaster;
+	use crate::types::{DynStore, DynStoreWrapper};
+	use crate::{NodeMetrics, PersistedNodeMetrics};
+
+	const EXTERNAL_DESCRIPTOR: &str = "wpkh(tprv8ZgxMBicQKsPdy6LMhUtFHAgpocR8GC6QmwMSFpZs7h6Eziw3SpThFfczTDh5rW2krkqffa11UpX3XkeTTB2FvzZKWXqPY54Y6Rq4AQ5R8L/84'/1'/0'/0/*)";
+	const INTERNAL_DESCRIPTOR: &str = "wpkh(tprv8ZgxMBicQKsPdy6LMhUtFHAgpocR8GC6QmwMSFpZs7h6Eziw3SpThFfczTDh5rW2krkqffa11UpX3XkeTTB2FvzZKWXqPY54Y6Rq4AQ5R8L/84'/1'/0'/1/*)";
+
+	/// A pass-through [`KVStore`] that parks writes to one namespace: a matching writer first
+	/// signals `parked`, then waits until the test drops its `gate` write guard. Writes to every
+	/// other namespace pass straight through.
+	#[derive(Clone)]
+	struct NamespaceGatedStore {
+		inner: Arc<InMemoryStore>,
+		gated_namespace: String,
+		parked: Arc<tokio::sync::Notify>,
+		gate: Arc<tokio::sync::RwLock<()>>,
+	}
+
+	impl NamespaceGatedStore {
+		fn new(gated_namespace: &str) -> Self {
+			Self {
+				inner: Arc::new(InMemoryStore::new()),
+				gated_namespace: gated_namespace.to_string(),
+				parked: Arc::new(tokio::sync::Notify::new()),
+				gate: Arc::new(tokio::sync::RwLock::new(())),
+			}
+		}
+	}
+
+	impl KVStore for NamespaceGatedStore {
+		fn read(
+			&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
+		) -> impl Future<Output = Result<Vec<u8>, io::Error>> + 'static + Send {
+			KVStore::read(&*self.inner, primary_namespace, secondary_namespace, key)
+		}
+
+		fn write(
+			&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec<u8>,
+		) -> impl Future<Output = Result<(), io::Error>> + 'static + Send {
+			let inner = Arc::clone(&self.inner);
+			let gated = primary_namespace == self.gated_namespace;
+			let parked = Arc::clone(&self.parked);
+			let gate = Arc::clone(&self.gate);
+			let primary_namespace = primary_namespace.to_string();
+			let secondary_namespace = secondary_namespace.to_string();
+			let key = key.to_string();
+			async move {
+				if gated {
+					parked.notify_one();
+					let _guard = gate.read().await;
+				}
+				KVStore::write(&*inner, &primary_namespace, &secondary_namespace, &key, buf).await
+			}
+		}
+
+		fn remove(
+			&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool,
+		) -> impl Future<Output = Result<(), io::Error>> + 'static + Send {
+			KVStore::remove(&*self.inner, primary_namespace, secondary_namespace, key, lazy)
+		}
+
+		fn list(
+			&self, primary_namespace: &str, secondary_namespace: &str,
+		) -> impl Future<Output = Result<Vec<String>, io::Error>> + 'static + Send {
+			KVStore::list(&*self.inner, primary_namespace, secondary_namespace)
+		}
+	}
+
+	impl PaginatedKVStore for NamespaceGatedStore {
+		fn list_paginated(
+			&self, primary_namespace: &str, secondary_namespace: &str,
+			page_token: Option<PageToken>,
+		) -> impl Future<Output = Result<PaginatedListResponse, io::Error>> + 'static + Send {
+			PaginatedKVStore::list_paginated(
+				&*self.inner,
+				primary_namespace,
+				secondary_namespace,
+				page_token,
+			)
+		}
+	}
+
+	/// Builds a [`Wallet`] whose every component is in-memory; nothing performs I/O at
+	/// construction, so wallet-sync and classification entry points can be driven directly.
+	async fn new_test_wallet(kv_store: Arc<DynStore>) -> Wallet {
+		let logger = Arc::new(Logger::new_log_facade());
+		let mut wallet_persister =
+			KVStoreWalletPersister::new(Arc::clone(&kv_store), Arc::clone(&logger));
+		let bdk_wallet = bdk_wallet::Wallet::create(EXTERNAL_DESCRIPTOR, INTERNAL_DESCRIPTOR)
+			.network(Network::Regtest)
+			.create_wallet_async(&mut wallet_persister)
+			.await
+			.unwrap();
+		let broadcaster = Arc::new(TransactionBroadcaster::new(Arc::clone(&logger)));
+		let fee_estimator = Arc::new(OnchainFeeEstimator::new());
+		let config = Arc::new(Config::default());
+		let node_metrics = Arc::new(PersistedNodeMetrics::new(NodeMetrics::default()));
+		let (chain_source, _) = ChainSource::new_esplora(
+			"http://127.0.0.1:1".to_string(),
+			HashMap::new(),
+			EsploraSyncConfig::default(),
+			Arc::clone(&fee_estimator),
+			Arc::clone(&broadcaster),
+			Arc::clone(&kv_store),
+			Arc::clone(&config),
+			Arc::clone(&logger),
+			node_metrics,
+		)
+		.unwrap();
+		let payment_store = Arc::new(PaymentStore::new(
+			Vec::new(),
+			PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE.to_string(),
+			PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE.to_string(),
+			Arc::clone(&kv_store),
+			Arc::clone(&logger),
+		));
+		let pending_payment_store = Arc::new(PendingPaymentStore::new(
+			Vec::new(),
+			PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE.to_string(),
+			PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE.to_string(),
+			Arc::clone(&kv_store),
+			Arc::clone(&logger),
+		));
+		let runtime = Arc::new(Runtime::new(Arc::clone(&logger)).unwrap());
+		Wallet::new(
+			bdk_wallet,
+			wallet_persister,
+			broadcaster,
+			fee_estimator,
+			Arc::new(chain_source),
+			payment_store,
+			runtime,
+			config,
+			logger,
+			pending_payment_store,
+		)
+	}
+
+	fn dummy_tx() -> Transaction {
+		Transaction {
+			version: bitcoin::transaction::Version::TWO,
+			lock_time: LockTime::ZERO,
+			input: Vec::new(),
+			output: Vec::new(),
+		}
+	}
+
+	fn confirmed_block_time(height: u32) -> ConfirmationBlockTime {
+		ConfirmationBlockTime {
+			block_id: BlockId { height, hash: bitcoin::BlockHash::from_byte_array([9u8; 32]) },
+			confirmation_time: 100,
+		}
+	}
+
+	fn interactive_funding_details(
+		id: PaymentId, txid: Txid, amount_msat: Option<u64>, fee_paid_msat: Option<u64>,
+	) -> PaymentDetails {
+		let kind = PaymentKind::Onchain {
+			txid,
+			status: ConfirmationStatus::Unconfirmed,
+			tx_type: Some(TransactionType::InteractiveFunding { channels: vec![] }),
+		};
+		PaymentDetails::new(
+			id,
+			kind,
+			amount_msat,
+			fee_paid_msat,
+			PaymentDirection::Outbound,
+			PaymentStatus::Pending,
+		)
+	}
+
+	fn onchain_details(txid: Txid, status: ConfirmationStatus) -> PaymentDetails {
+		PaymentDetails::new(
+			PaymentId([42u8; 32]),
+			PaymentKind::Onchain { txid, status, tx_type: None },
+			Some(1_000_000),
+			Some(500),
+			PaymentDirection::Outbound,
+			PaymentStatus::Pending,
+		)
+	}
+
+	fn confirmed_status() -> ConfirmationStatus {
+		ConfirmationStatus::Confirmed {
+			block_hash: bitcoin::BlockHash::from_byte_array([8u8; 32]),
+			height: 100,
+			timestamp: 1,
+		}
+	}
+
+	#[test]
+	fn funding_reclassification_update_substitutes_the_confirmed_candidate() {
+		let confirmed_txid = Txid::from_byte_array([1u8; 32]);
+		let active_txid = Txid::from_byte_array([2u8; 32]);
+		let candidates = vec![
+			FundingTxCandidate {
+				txid: confirmed_txid,
+				amount_msat: Some(2_000_000),
+				fee_paid_msat: Some(999),
+			},
+			FundingTxCandidate {
+				txid: active_txid,
+				amount_msat: Some(1_000_000),
+				fee_paid_msat: Some(500),
+			},
+		];
+		let details = onchain_details(active_txid, ConfirmationStatus::Unconfirmed);
+
+		// The record confirmed an earlier candidate: the update reports that candidate, not the
+		// active one.
+		let current = onchain_details(confirmed_txid, confirmed_status());
+		let update = funding_reclassification_update(details.clone(), &candidates, Some(&current));
+		assert_eq!(update.txid, Some(confirmed_txid));
+		assert_eq!(update.amount_msat, Some(Some(2_000_000)));
+		assert_eq!(update.fee_paid_msat, Some(Some(999)));
+
+		// A confirmed candidate we did not contribute to still substitutes, with empty figures —
+		// the same figures a confirmation arriving after classification would report.
+		let uncontributed = vec![FundingTxCandidate {
+			txid: confirmed_txid,
+			amount_msat: None,
+			fee_paid_msat: None,
+		}];
+		let update =
+			funding_reclassification_update(details.clone(), &uncontributed, Some(&current));
+		assert_eq!(update.txid, Some(confirmed_txid));
+		assert_eq!(update.amount_msat, Some(None));
+		assert_eq!(update.fee_paid_msat, Some(None));
+	}
+
+	#[test]
+	fn funding_reclassification_update_keeps_the_active_candidate() {
+		let active_txid = Txid::from_byte_array([2u8; 32]);
+		let candidates = vec![FundingTxCandidate {
+			txid: active_txid,
+			amount_msat: Some(1_000_000),
+			fee_paid_msat: Some(500),
+		}];
+		let details = onchain_details(active_txid, ConfirmationStatus::Unconfirmed);
+
+		// No record yet: the update describes the active candidate.
+		let update = funding_reclassification_update(details.clone(), &candidates, None);
+		assert_eq!(update.txid, Some(active_txid));
+		assert_eq!(update.amount_msat, Some(Some(1_000_000)));
+
+		// An unconfirmed record: still the active candidate (RBF rotation).
+		let unconfirmed =
+			onchain_details(Txid::from_byte_array([1u8; 32]), ConfirmationStatus::Unconfirmed);
+		let update =
+			funding_reclassification_update(details.clone(), &candidates, Some(&unconfirmed));
+		assert_eq!(update.txid, Some(active_txid));
+
+		// The record confirmed the active candidate itself: nothing to substitute.
+		let current = onchain_details(active_txid, confirmed_status());
+		let update = funding_reclassification_update(details.clone(), &candidates, Some(&current));
+		assert_eq!(update.txid, Some(active_txid));
+		assert_eq!(update.amount_msat, Some(Some(1_000_000)));
+
+		// A confirmed txid outside the candidate history (e.g. the record is an unrelated
+		// same-id payment): fall back to the active candidate; `PaymentDetails::update` keeps
+		// the confirmed figures in place on mismatch.
+		let foreign = onchain_details(Txid::from_byte_array([9u8; 32]), confirmed_status());
+		let update = funding_reclassification_update(details, &candidates, Some(&foreign));
+		assert_eq!(update.txid, Some(active_txid));
+	}
+
+	/// Graduation must decide from the live record and write only the status: a pending-store
+	/// snapshot taken before a concurrent classification landed must not roll the record's
+	/// figures back when the payment graduates to `Succeeded`.
+	#[tokio::test]
+	async fn graduation_preserves_classified_figures() {
+		let store: Arc<DynStore> = Arc::new(DynStoreWrapper(InMemoryStore::new()));
+		let wallet = new_test_wallet(store).await;
+
+		let txid = Txid::from_byte_array([4u8; 32]);
+		let payment_id = PaymentId(txid.to_byte_array());
+		let confirmed = ConfirmationStatus::Confirmed {
+			block_hash: bitcoin::BlockHash::from_byte_array([9u8; 32]),
+			height: 5,
+			timestamp: 100,
+		};
+		let tx_type = Some(TransactionType::InteractiveFunding { channels: vec![] });
+
+		// The live record carries the classification: contribution-derived figures, confirmed.
+		let mut recorded =
+			interactive_funding_details(payment_id, txid, Some(2_000_000), Some(999));
+		recorded.kind = PaymentKind::Onchain { txid, status: confirmed, tx_type: tx_type.clone() };
+		recorded.latest_update_timestamp = 0;
+		wallet.payment_store.insert_or_update(recorded).await.unwrap();
+
+		// The pending entry embeds a stale snapshot: wallet-derived figures recorded before the
+		// classification above landed.
+		let mut stale = interactive_funding_details(payment_id, txid, Some(0), Some(0));
+		stale.kind = PaymentKind::Onchain { txid, status: confirmed, tx_type };
+		let entry = PendingPaymentDetails::new(stale, Vec::new(), Vec::new());
+		wallet.pending_payment_store.insert_or_update(entry).await.unwrap();
+
+		let block_id =
+			|height| BlockId { height, hash: bitcoin::BlockHash::from_byte_array([7u8; 32]) };
+		let event = WalletEvent::ChainTipChanged { old_tip: block_id(9), new_tip: block_id(10) };
+		wallet.update_payment_store(vec![event]).await.unwrap();
+
+		let payment = wallet.payment_store.get(&payment_id).unwrap();
+		assert_eq!(payment.status, PaymentStatus::Succeeded);
+		assert_eq!(
+			payment.amount_msat,
+			Some(2_000_000),
+			"graduation must not roll figures back to the snapshot's"
+		);
+		assert_eq!(payment.fee_paid_msat, Some(999));
+		assert!(payment.latest_update_timestamp > 0, "the graduation write must timestamp");
+		assert!(wallet.pending_payment_store.get(&payment_id).is_none());
+	}
+
+	/// When the live record has diverged from the pending-store snapshot — here the snapshot
+	/// says Confirmed at graduation depth while the record says Unconfirmed — graduation must
+	/// decline and keep the entry rather than force-writing `Succeeded` from stale state. The
+	/// seeded divergence is synthetic (no current production writer downgrades a record's
+	/// confirmation); the test pins the hardening that comes with deciding from the live record.
+	#[tokio::test]
+	async fn graduation_declines_on_diverged_record() {
+		let store: Arc<DynStore> = Arc::new(DynStoreWrapper(InMemoryStore::new()));
+		let wallet = new_test_wallet(store).await;
+
+		let txid = Txid::from_byte_array([5u8; 32]);
+		let payment_id = PaymentId(txid.to_byte_array());
+		let confirmed = ConfirmationStatus::Confirmed {
+			block_hash: bitcoin::BlockHash::from_byte_array([9u8; 32]),
+			height: 5,
+			timestamp: 100,
+		};
+
+		// The live record is Unconfirmed...
+		let recorded = interactive_funding_details(payment_id, txid, Some(2_000_000), Some(999));
+		wallet.payment_store.insert_or_update(recorded).await.unwrap();
+
+		// ...while the pending entry's snapshot claims a graduation-deep confirmation.
+		let mut snapshot =
+			interactive_funding_details(payment_id, txid, Some(2_000_000), Some(999));
+		snapshot.kind = PaymentKind::Onchain {
+			txid,
+			status: confirmed,
+			tx_type: Some(TransactionType::InteractiveFunding { channels: vec![] }),
+		};
+		let entry = PendingPaymentDetails::new(snapshot, Vec::new(), Vec::new());
+		wallet.pending_payment_store.insert_or_update(entry).await.unwrap();
+
+		let block_id =
+			|height| BlockId { height, hash: bitcoin::BlockHash::from_byte_array([7u8; 32]) };
+		let event = WalletEvent::ChainTipChanged { old_tip: block_id(9), new_tip: block_id(10) };
+		wallet.update_payment_store(vec![event]).await.unwrap();
+
+		let payment = wallet.payment_store.get(&payment_id).unwrap();
+		assert_eq!(
+			payment.status,
+			PaymentStatus::Pending,
+			"a diverged snapshot must not force-graduate the record"
+		);
+		assert!(matches!(
+			payment.kind,
+			PaymentKind::Onchain { status: ConfirmationStatus::Unconfirmed, .. }
+		));
+		assert!(
+			wallet.pending_payment_store.get(&payment_id).is_some(),
+			"the entry must survive for future events to drive"
+		);
+	}
+
+	/// A middle RBF candidate must map back to the funding record: it is neither the record's
+	/// id (derived from the first candidate), nor its current txid (the active candidate), nor
+	/// in `conflicting_txids` (it never got a `TxReplaced` event of its own).
+	#[tokio::test]
+	async fn find_payment_by_txid_maps_candidate_txids() {
+		let store: Arc<DynStore> = Arc::new(DynStoreWrapper(InMemoryStore::new()));
+		let wallet = new_test_wallet(store).await;
+
+		let txid1 = Txid::from_byte_array([1u8; 32]);
+		let txid2 = Txid::from_byte_array([2u8; 32]);
+		let txid3 = Txid::from_byte_array([3u8; 32]);
+		let payment_id = PaymentId(txid1.to_byte_array());
+		let candidates = vec![
+			FundingTxCandidate {
+				txid: txid1,
+				amount_msat: Some(1_000_000),
+				fee_paid_msat: Some(500),
+			},
+			FundingTxCandidate {
+				txid: txid2,
+				amount_msat: Some(1_000_000),
+				fee_paid_msat: Some(600),
+			},
+			FundingTxCandidate {
+				txid: txid3,
+				amount_msat: Some(1_000_000),
+				fee_paid_msat: Some(700),
+			},
+		];
+		let details = interactive_funding_details(payment_id, txid3, Some(1_000_000), Some(700));
+		let entry = PendingPaymentDetails::new(details, Vec::new(), candidates);
+		wallet.pending_payment_store.insert_or_update(entry).await.unwrap();
+
+		// The first candidate resolves via the txid-derived id and the active candidate via the
+		// record's current txid; the middle one must resolve through the candidate history.
+		assert_eq!(wallet.find_payment_by_txid(txid1), Some(payment_id));
+		assert_eq!(wallet.find_payment_by_txid(txid3), Some(payment_id));
+		assert_eq!(wallet.find_payment_by_txid(txid2), Some(payment_id));
+	}
+
+	/// Barrier test, classification-first ordering: wallet sync's confirmation handling must
+	/// wait for classification's two-store write pair. Classification is parked between its
+	/// payment-store and pending-store writes (the torn window) and only then is the
+	/// confirmation of the replacement candidate dispatched; unless the sync arm holds the
+	/// cross-store lock from payment-id resolution onwards, it resolves the id against the
+	/// still-missing pending index and mints a duplicate record keyed by the event txid.
+	#[tokio::test]
+	async fn funding_confirmation_waits_for_classification() {
+		let gated = NamespaceGatedStore::new(PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE);
+		let store: Arc<DynStore> = Arc::new(DynStoreWrapper(gated.clone()));
+		let wallet = Arc::new(new_test_wallet(store).await);
+
+		let txid1 = Txid::from_byte_array([1u8; 32]);
+		let txid2 = Txid::from_byte_array([2u8; 32]);
+		let payment_id = PaymentId(txid1.to_byte_array());
+		let candidates = vec![
+			FundingTxCandidate {
+				txid: txid1,
+				amount_msat: Some(1_000_000),
+				fee_paid_msat: Some(500),
+			},
+			FundingTxCandidate {
+				txid: txid2,
+				amount_msat: Some(2_000_000),
+				fee_paid_msat: Some(999),
+			},
+		];
+		let details = interactive_funding_details(payment_id, txid2, Some(2_000_000), Some(999));
+
+		// Hold the gate so classification parks on its pending-store write: the payment record
+		// is persisted, the pending entry is not — the torn window a concurrent confirmation
+		// must not observe.
+		let gate_guard = gated.gate.write().await;
+		let classification = tokio::spawn({
+			let wallet = Arc::clone(&wallet);
+			let candidates = candidates.clone();
+			async move { wallet.persist_funding_payment(details, candidates).await }
+		});
+		gated.parked.notified().await;
+
+		// Only now dispatch the confirmation of the candidate that won.
+		let event = WalletEvent::TxConfirmed {
+			txid: txid2,
+			tx: Arc::new(dummy_tx()),
+			block_time: confirmed_block_time(5),
+			old_block_time: None,
+		};
+		let sync = tokio::spawn({
+			let wallet = Arc::clone(&wallet);
+			async move { wallet.update_payment_store(vec![event]).await }
+		});
+
+		// Liveness sanity only (both pre- and post-fix stall here): while classification is
+		// parked, no second record may have been committed.
+		tokio::time::sleep(Duration::from_millis(250)).await;
+		assert!(wallet.payment_store.list_filter(|_| true).len() <= 1);
+
+		drop(gate_guard);
+		classification.await.unwrap().unwrap();
+		sync.await.unwrap().unwrap();
+
+		// Both writers converge on the classified record: the confirmation refreshes it in
+		// place with the confirmed candidate's figures rather than minting a second record
+		// keyed by the event txid.
+		let payments = wallet.payment_store.list_filter(|_| true);
+		assert_eq!(payments.len(), 1, "the confirmation must not mint a duplicate record");
+		let payment = &payments[0];
+		assert_eq!(payment.id, payment_id);
+		assert_eq!(payment.amount_msat, Some(2_000_000));
+		assert_eq!(payment.fee_paid_msat, Some(999));
+		match &payment.kind {
+			PaymentKind::Onchain {
+				txid,
+				status: ConfirmationStatus::Confirmed { .. },
+				tx_type: Some(TransactionType::InteractiveFunding { .. }),
+			} => assert_eq!(*txid, txid2),
+			kind => panic!("unexpected kind {:?}", kind),
+		}
+	}
+
+	/// Barrier test, sync-first ordering: classification must wait for wallet sync's complete
+	/// decision-plus-write sequence. Wallet sync is parked inside its generic-fallback window —
+	/// past the funding-status check that found no record, before its writes — by holding the
+	/// BDK wallet lock the fallback needs. Unless the sync arm holds the cross-store lock
+	/// across that window, classification lands in between and the fallback's stale merge
+	/// overwrites the contribution-derived figures with wallet-derived ones.
+	#[tokio::test(flavor = "multi_thread")]
+	async fn funding_classification_waits_for_wallet_sync() {
+		let store: Arc<DynStore> = Arc::new(DynStoreWrapper(InMemoryStore::new()));
+		let wallet = Arc::new(new_test_wallet(store).await);
+
+		let txid = Txid::from_byte_array([3u8; 32]);
+		let payment_id = PaymentId(txid.to_byte_array());
+		let candidates = vec![FundingTxCandidate {
+			txid,
+			amount_msat: Some(1_000_000),
+			fee_paid_msat: Some(500),
+		}];
+		let details = interactive_funding_details(payment_id, txid, Some(1_000_000), Some(500));
+
+		// Park wallet sync inside its fallback window: the TxUnconfirmed arm reads no wallet
+		// state before that point, so it passes the funding-status check (no record exists yet)
+		// and then blocks on the wallet lock held here. The sleeps give the tasks time to reach
+		// their parking spots; they make the pre-fix failure deterministic, while the fixed
+		// code converges to the same final state under any arrival order.
+		let inner_guard = wallet.inner.lock().unwrap();
+		let sync = tokio::spawn({
+			let wallet = Arc::clone(&wallet);
+			let event =
+				WalletEvent::TxUnconfirmed { txid, tx: Arc::new(dummy_tx()), old_block_time: None };
+			async move { wallet.update_payment_store(vec![event]).await }
+		});
+		tokio::time::sleep(Duration::from_millis(250)).await;
+
+		let classification = tokio::spawn({
+			let wallet = Arc::clone(&wallet);
+			let candidates = candidates.clone();
+			async move { wallet.persist_funding_payment(details, candidates).await }
+		});
+		tokio::time::sleep(Duration::from_millis(250)).await;
+
+		drop(inner_guard);
+		sync.await.unwrap().unwrap();
+		classification.await.unwrap().unwrap();
+
+		// Both writers converge on one record carrying the classification: the generic
+		// fallback must not clobber the contribution-derived figures with its wallet-derived
+		// view of the transaction.
+		let payments = wallet.payment_store.list_filter(|_| true);
+		assert_eq!(payments.len(), 1);
+		let payment = &payments[0];
+		assert_eq!(payment.id, payment_id);
+		assert_eq!(
+			payment.amount_msat,
+			Some(1_000_000),
+			"wallet sync's fallback must not overwrite contribution figures"
+		);
+		assert_eq!(payment.fee_paid_msat, Some(500));
+		assert!(matches!(
+			&payment.kind,
+			PaymentKind::Onchain { tx_type: Some(TransactionType::InteractiveFunding { .. }), .. }
+		));
+	}
+}