diff --git a/packages/rs-drive/tests/query_tests.rs b/packages/rs-drive/tests/query_tests.rs
index e8a7ba50a35..ce5927997ea 100644
--- a/packages/rs-drive/tests/query_tests.rs
+++ b/packages/rs-drive/tests/query_tests.rs
@@ -9853,3 +9853,236 @@ mod gapped_index_query_tests {
         );
     }
 }
+
+#[cfg(feature = "server")]
+#[cfg(test)]
+mod dpns_identity_records_paging {
+    //! Pins the paging shape the SDK uses to list every DPNS name of an
+    //! identity (`Sdk::get_all_dpns_usernames_by_identity`): an equality on
+    //! `records.identity` (the non-unique `identityId` index), no `orderBy`,
+    //! continued with `startAfter` the previous page's last document id. The
+    //! SDK treats a short page as "the whole set", so a continuation that
+    //! drops the rest of the equality bucket would make it prune names.
+
+    use super::*;
+    use dpp::data_contracts::SystemDataContract;
+    use dpp::system_data_contracts::load_system_data_contract;
+    use drive::util::object_size_info::DocumentInfo;
+    use drive::util::test_helpers::setup::setup_drive_with_initial_state_structure;
+
+    const IDENTITY: [u8; 32] = [0x11; 32];
+    const OTHER_IDENTITY: [u8; 32] = [0x22; 32];
+
+    /// DPNS contract plus three domains resolving to `IDENTITY` (ids 1, 2, 3)
+    /// and one resolving elsewhere (id 4), all written at `platform_version`.
+    fn setup(platform_version: &PlatformVersion) -> (Drive, DataContract) {
+        let drive = setup_drive_with_initial_state_structure(Some(platform_version));
+        let contract = load_system_data_contract(SystemDataContract::DPNS, platform_version)
+            .expect("expected to load the DPNS contract");
+        drive
+            .apply_contract(
+                &contract,
+                BlockInfo::default(),
+                true,
+                None,
+                None,
+                platform_version,
+            )
+            .expect("expected to apply the DPNS contract");
+        let document_type = contract
+            .document_type_for_name("domain")
+            .expect("expected a domain document type");
+        for (id, identity) in [
+            (1u8, IDENTITY),
+            (2, IDENTITY),
+            (3, IDENTITY),
+            (4, OTHER_IDENTITY),
+        ] {
+            let label = format!("paging-test-name-number-{id}");
+            let mut document = document_type
+                .create_document_from_data(
+                    platform_value!({
+                        "label": label.clone(),
+                        "normalizedLabel": label,
+                        "parentDomainName": "dash",
+                        "normalizedParentDomainName": "dash",
+                        "preorderSalt": Bytes32::new([id; 32]),
+                        "records": {"identity": Identifier::from(identity)},
+                        "subdomainRules": {"allowSubdomains": false},
+                    }),
+                    Identifier::from(identity),
+                    1,
+                    1,
+                    [id; 32],
+                    platform_version,
+                )
+                .expect("expected to create a domain document");
+            document.set_id(Identifier::from([id; 32]));
+            drive
+                .add_document_for_contract(
+                    DocumentAndContractInfo {
+                        owned_document_info: OwnedDocumentInfo {
+                            document_info: DocumentInfo::DocumentRefInfo((&document, None)),
+                            owner_id: None,
+                        },
+                        contract: &contract,
+                        document_type,
+                    },
+                    true,
+                    BlockInfo::default(),
+                    true,
+                    None,
+                    platform_version,
+                    None,
+                )
+                .expect("expected to insert a domain document");
+        }
+        (drive, contract)
+    }
+
+    /// One page of `IDENTITY`'s domains through the raw path, the proved
+    /// path and the client-side proof verification the SDK runs; all three
+    /// must agree. Returns the document ids of the page.
+    fn page(
+        drive: &Drive,
+        contract: &DataContract,
+        start_after: Option<u8>,
+        platform_version: &PlatformVersion,
+    ) -> Vec<Identifier> {
+        let document_type = contract
+            .document_type_for_name("domain")
+            .expect("expected a domain document type");
+        let mut query_value = json!({
+            "where": [
+                ["records.identity", "==", bs58::encode(IDENTITY).into_string()]
+            ],
+            "limit": 2,
+        });
+        if let Some(id) = start_after {
+            query_value["startAfter"] = json!(bs58::encode([id; 32]).into_string());
+        }
+        let query_bytes = cbor_serializer::serializable_value_to_cbor(&query_value, None)
+            .expect("expected to serialize the query");
+        let query = || {
+            DriveDocumentQuery::from_cbor(
+                &query_bytes,
+                contract,
+                document_type,
+                &drive.config,
+                platform_version,
+            )
+            .expect("expected the query to be built")
+        };
+        let ids_of = |results: Vec<Vec<u8>>| -> Vec<Identifier> {
+            results
+                .iter()
+                .map(|bytes| {
+                    Document::from_bytes(bytes, document_type, platform_version)
+                        .expect("expected to deserialize a result")
+                        .id()
+                })
+                .collect()
+        };
+
+        let (raw, _, _) = query()
+            .execute_raw_results_no_proof(drive, None, None, platform_version)
+            .expect("expected the raw query to execute");
+        let raw = ids_of(raw);
+
+        let (_, proved, _) = query()
+            .execute_with_proof_only_get_elements(drive, None, None, platform_version)
+            .expect("expected the proved query to execute");
+        assert_eq!(raw, ids_of(proved), "raw and proved pages must agree");
+
+        let (proof, _) = query()
+            .execute_with_proof(drive, None, None, platform_version)
+            .expect("expected a proof");
+        let (_, verified) = query()
+            .verify_proof(&proof, platform_version)
+            .expect("expected the proof to verify");
+        let verified: Vec<Identifier> = verified.iter().map(|document| document.id()).collect();
+        assert_eq!(raw, verified, "the verified page must match the served one");
+
+        raw
+    }
+
+    fn assert_first_page(
+        drive: &Drive,
+        contract: &DataContract,
+        platform_version: &PlatformVersion,
+    ) {
+        assert_eq!(
+            page(drive, contract, None, platform_version),
+            vec![Identifier::from([1; 32]), Identifier::from([2; 32])],
+            "protocol version {}: first page",
+            platform_version.protocol_version
+        );
+    }
+
+    #[test]
+    fn should_continue_after_a_document_inside_an_equality_bucket_at_latest_protocol_version() {
+        let platform_version = PlatformVersion::latest();
+        assert_eq!(
+            platform_version
+                .drive
+                .methods
+                .document
+                .query
+                .non_primary_key_path_query,
+            1
+        );
+        let (drive, contract) = setup(platform_version);
+
+        assert_first_page(&drive, &contract, platform_version);
+        assert_eq!(
+            page(&drive, &contract, Some(2), platform_version),
+            vec![Identifier::from([3; 32])],
+            "startAfter the second document must return the third"
+        );
+        assert_eq!(
+            page(&drive, &contract, Some(1), platform_version),
+            vec![Identifier::from([2; 32]), Identifier::from([3; 32])],
+            "startAfter the first document must return the rest of the bucket"
+        );
+        assert_eq!(
+            page(&drive, &contract, Some(3), platform_version),
+            Vec::<Identifier>::new(),
+            "startAfter the last document is empty"
+        );
+    }
+
+    /// The v0 lowering (protocol versions <= 13) hands an exclusive cursor
+    /// to the equality clause's own key: `WhereClause::to_path_query`'s
+    /// `Equal` arm drops the key when it equals the cursor document's value
+    /// and the cursor is not included, which on a single-property index is
+    /// always the case. Every continuation inside the bucket therefore comes
+    /// back empty — served, proved and verified alike — so a client cannot
+    /// tell "no more documents" from "the cursor dropped them", and a full
+    /// first page is the most a v0 network can be asked for.
+    ///
+    /// On chain behavior: never edit these expectations; protocol version 14
+    /// (`non_primary_key_path_query: 1`) is the fix.
+    #[test]
+    fn should_drop_the_rest_of_an_equality_bucket_after_start_after_frozen_at_protocol_v13() {
+        let platform_version = PlatformVersion::get(13).expect("expected protocol version 13");
+        assert_eq!(
+            platform_version
+                .drive
+                .methods
+                .document
+                .query
+                .non_primary_key_path_query,
+            0
+        );
+        let (drive, contract) = setup(platform_version);
+
+        assert_first_page(&drive, &contract, platform_version);
+        for cursor in [1u8, 2, 3] {
+            assert_eq!(
+                page(&drive, &contract, Some(cursor), platform_version),
+                Vec::<Identifier>::new(),
+                "v0 drops the whole bucket after startAfter document {cursor}"
+            );
+        }
+    }
+}
diff --git a/packages/rs-platform-wallet-ffi/src/persistence.rs b/packages/rs-platform-wallet-ffi/src/persistence.rs
index d95a1f31db8..c66d0f327e5 100644
--- a/packages/rs-platform-wallet-ffi/src/persistence.rs
+++ b/packages/rs-platform-wallet-ffi/src/persistence.rs
@@ -10921,4 +10921,40 @@ mod tests {
             "every emitted marked-used address must carry used == true"
         );
     }
+
+    /// A cold restore must hand every persisted owned DPNS label to the
+    /// in-memory identity, so a later capped fetch only adds to it and no
+    /// snapshot emitted before the next complete fetch drops an owned pick
+    /// that lies outside the fetched prefix.
+    #[test]
+    fn restored_identity_carries_its_persisted_dpns_names() {
+        use super::build_wallet_identity_bucket;
+        use crate::wallet_restore_types::{IdentityRestoreEntryFFI, WalletRestoreEntryFFI};
+        use std::ffi::CString;
+        use std::os::raw::c_char;
+
+        let labels: Vec<CString> = ["alice", "bob", "carol"]
+            .iter()
+            .map(|l| CString::new(*l).unwrap())
+            .collect();
+        let pointers: Vec<*const c_char> = labels.iter().map(|l| l.as_ptr()).collect();
+        let mut identity: IdentityRestoreEntryFFI = unsafe { std::mem::zeroed() };
+        identity.identity_id = [7; 32];
+        identity.dpns_names = pointers.as_ptr();
+        identity.dpns_names_count = pointers.len();
+        let entry = WalletRestoreEntryFFI {
+            wallet_id: [42; 32],
+            identities: &identity,
+            identities_count: 1,
+            ..Default::default()
+        };
+
+        let bucket = build_wallet_identity_bucket(&entry).expect("identity must restore");
+        let restored: Vec<&str> = bucket[&0]
+            .dpns_names
+            .iter()
+            .map(|n| n.label.as_str())
+            .collect();
+        assert_eq!(restored, ["alice", "bob", "carol"]);
+    }
 }
diff --git a/packages/rs-platform-wallet/src/lib.rs b/packages/rs-platform-wallet/src/lib.rs
index 2fedc74a1a8..f6af75092fd 100644
--- a/packages/rs-platform-wallet/src/lib.rs
+++ b/packages/rs-platform-wallet/src/lib.rs
@@ -82,7 +82,7 @@ pub use wallet::identity::{
     calculate_account_reference, derive_auto_accept_private_key, derive_contact_payment_address,
     derive_contact_payment_addresses, derive_contact_xpub, pubkey_binds_expected_key_data,
     unmask_account_reference, BlockTime, ContactProfileEntry, ContactRequest, ContactXpubData,
-    DashPayProfile, DashPayState, DpnsNameInfo, EstablishedContact, IdentityLocation,
+    DashPayProfile, DashPayState, DpnsFetch, DpnsNameInfo, EstablishedContact, IdentityLocation,
     IdentityManager, IdentityStatus, KeyStorage, ManagedIdentity, PrivateKeyData, ProfileUpdate,
     RegistrationIndex, DEFAULT_CONTACT_GAP_LIMIT,
 };
diff --git a/packages/rs-platform-wallet/src/wallet/identity/mod.rs b/packages/rs-platform-wallet/src/wallet/identity/mod.rs
index 54b7b8bb300..01079577e1b 100644
--- a/packages/rs-platform-wallet/src/wallet/identity/mod.rs
+++ b/packages/rs-platform-wallet/src/wallet/identity/mod.rs
@@ -35,7 +35,7 @@ pub use state::{
 };
 pub use types::dashpay::profile::{calculate_avatar_hash, calculate_dhash_fingerprint};
 pub use types::{
-    ContactProfileEntry, ContactRequest, DashPayProfile, DashpayAddressMatch, DpnsNameInfo,
-    EstablishedContact, IdentityStatus, KeyStorage, PaymentDirection, PaymentEntry, PaymentStatus,
-    PrivateKeyData, ProfileUpdate,
+    ContactProfileEntry, ContactRequest, DashPayProfile, DashpayAddressMatch, DpnsFetch,
+    DpnsNameInfo, EstablishedContact, IdentityStatus, KeyStorage, PaymentDirection, PaymentEntry,
+    PaymentStatus, PrivateKeyData, ProfileUpdate,
 };
diff --git a/packages/rs-platform-wallet/src/wallet/identity/network/discovery.rs b/packages/rs-platform-wallet/src/wallet/identity/network/discovery.rs
index 6117a170f7b..bb8231dcf30 100644
--- a/packages/rs-platform-wallet/src/wallet/identity/network/discovery.rs
+++ b/packages/rs-platform-wallet/src/wallet/identity/network/discovery.rs
@@ -335,7 +335,6 @@ impl IdentityWallet {
         enrichment_deadline: Option<std::time::Instant>,
     ) -> Result<Vec<Identity>, PlatformWalletError> {
         use super::identity_handle::{derive_identity_auth_key_hash_from_master, MASTER_KEY_INDEX};
-        use crate::wallet::identity::state::managed_identity::key_storage::DpnsNameInfo;
         use crate::wallet::identity::state::managed_identity::key_storage::IdentityStatus;
         use dash_sdk::platform::types::identity::PublicKeyHash;
         use dash_sdk::platform::Fetch;
@@ -580,12 +579,8 @@ impl IdentityWallet {
                 break;
             }
             let identity_id = identity.id();
-            match self
-                .sdk
-                .get_dpns_usernames_by_identity(identity_id, None)
-                .await
-            {
-                Ok(usernames) => {
+            match self.fetch_owned_dpns_names(identity_id, None).await {
+                Ok((names, fetch)) => {
                     let mut wm_guard = self.wallet_manager.write().await;
                     let info_guard =
                         wm_guard
@@ -599,15 +594,7 @@ impl IdentityWallet {
                         .identity_manager
                         .managed_identity_mut(&identity_id)
                     {
-                        for username in usernames {
-                            managed.add_dpns_name(
-                                DpnsNameInfo {
-                                    label: username.label,
-                                    acquired_at: None,
-                                },
-                                &self.persister,
-                            );
-                        }
+                        managed.apply_fetched_dpns_names(names, fetch, &self.persister);
                     }
                 }
                 Err(e) => {
diff --git a/packages/rs-platform-wallet/src/wallet/identity/network/dpns.rs b/packages/rs-platform-wallet/src/wallet/identity/network/dpns.rs
index 1629bf2149e..fff5c8ba2f3 100644
--- a/packages/rs-platform-wallet/src/wallet/identity/network/dpns.rs
+++ b/packages/rs-platform-wallet/src/wallet/identity/network/dpns.rs
@@ -18,7 +18,9 @@ use dpp::address_funds::AddressWitness;
 use dpp::platform_value::BinaryData;
 
 use crate::error::PlatformWalletError;
-use crate::wallet::identity::types::key_storage::DpnsNameInfo;
+use crate::wallet::identity::types::key_storage::{DpnsFetch, DpnsNameInfo};
+
+use super::{DPNS_USERNAMES_MAX_PAGES, DPNS_USERNAMES_PAGE_LIMIT};
 
 use super::*;
 
@@ -317,13 +319,53 @@ impl IdentityWallet {
         })
     }
 
-    /// Fetch all DPNS usernames owned by `identity_id` from Platform
-    /// and merge them into the local
+    /// Every DPNS username `identity_id` owns on Platform, paged, as
+    /// [`DpnsNameInfo`]s stamped with `acquired_at`, and whether that is the
+    /// whole owned set ([`DpnsFetch::Complete`]) or only a lower bound — see
+    /// [`Sdk::get_all_dpns_usernames_by_identity`] for when paging can prove
+    /// completeness. Every username fetch that feeds
+    /// `ManagedIdentity::apply_fetched_dpns_names` goes through here.
+    pub(super) async fn fetch_owned_dpns_names(
+        &self,
+        identity_id: Identifier,
+        acquired_at: Option<u64>,
+    ) -> Result<(Vec<DpnsNameInfo>, DpnsFetch), dash_sdk::Error> {
+        let (usernames, complete) = self
+            .sdk
+            .get_all_dpns_usernames_by_identity(
+                identity_id,
+                DPNS_USERNAMES_PAGE_LIMIT,
+                DPNS_USERNAMES_MAX_PAGES,
+            )
+            .await?;
+        let names = usernames
+            .into_iter()
+            .map(|username| DpnsNameInfo {
+                label: username.label,
+                acquired_at,
+            })
+            .collect();
+        let fetch = if complete {
+            DpnsFetch::Complete
+        } else {
+            DpnsFetch::Partial
+        };
+        Ok((names, fetch))
+    }
+
+    /// Fetch the DPNS usernames owned by `identity_id` from Platform
+    /// and reconcile the local
     /// [`ManagedIdentity.dpns_names`](crate::wallet::identity::ManagedIdentity)
-    /// cache.
+    /// cache with them.
     ///
-    /// Skips labels that are already in the cache, so repeated syncs
-    /// don't emit duplicate entries. New labels get an
+    /// The query pages until a page comes back short. For an identity the
+    /// wallet only watches, that complete owned set replaces the cache (names
+    /// that left drop out); a wallet-owned identity only gains labels, since
+    /// the marketplace sweep owns its departures. If the page bound is
+    /// reached first, or the platform version cannot continue a full page
+    /// (protocol version <= 13), the result is only a lower bound and is
+    /// merged — see `ManagedIdentity::apply_fetched_dpns_names`. Known labels keep their
+    /// timestamp; new labels get an
     /// `acquired_at` timestamp of best-effort wall-clock millis —
     /// DPNS documents carry their own `$createdAt` but
     /// `DpnsUsername` doesn't surface it on the query result today.
@@ -339,9 +381,12 @@ impl IdentityWallet {
         &self,
         identity_id: &Identifier,
     ) -> Result<u32, PlatformWalletError> {
-        let usernames = self
-            .sdk
-            .get_dpns_usernames_by_identity(*identity_id, None)
+        let acquired_at = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .map(|d| d.as_millis() as u64)
+            .ok();
+        let (names, fetch) = self
+            .fetch_owned_dpns_names(*identity_id, acquired_at)
             .await
             .map_err(|e| {
                 PlatformWalletError::InvalidIdentityData(format!(
@@ -349,16 +394,7 @@ impl IdentityWallet {
                 ))
             })?;
 
-        if usernames.is_empty() {
-            return Ok(0);
-        }
-
-        let acquired_at = std::time::SystemTime::now()
-            .duration_since(std::time::UNIX_EPOCH)
-            .map(|d| d.as_millis() as u64)
-            .ok();
-
-        let mut added = 0u32;
+        let added;
         {
             let mut wm = self.wallet_manager.write().await;
             let Some(info) = wm.get_wallet_info_mut(&self.wallet_id) else {
@@ -367,23 +403,8 @@ impl IdentityWallet {
             let Some(managed) = info.identity_manager.managed_identity_mut(identity_id) else {
                 return Ok(0);
             };
-            for username in usernames {
-                if managed
-                    .dpns_names
-                    .iter()
-                    .any(|existing| existing.label == username.label)
-                {
-                    continue;
-                }
-                managed.add_dpns_name(
-                    DpnsNameInfo {
-                        label: username.label,
-                        acquired_at,
-                    },
-                    &self.persister,
-                );
-                added += 1;
-            }
+            // One snapshot for the whole fetch — see `apply_fetched_dpns_names`.
+            added = managed.apply_fetched_dpns_names(names, fetch, &self.persister);
         }
         Ok(added)
     }
diff --git a/packages/rs-platform-wallet/src/wallet/identity/network/dpns_marketplace.rs b/packages/rs-platform-wallet/src/wallet/identity/network/dpns_marketplace.rs
index b7884d4b45e..46fab23f519 100644
--- a/packages/rs-platform-wallet/src/wallet/identity/network/dpns_marketplace.rs
+++ b/packages/rs-platform-wallet/src/wallet/identity/network/dpns_marketplace.rs
@@ -313,6 +313,53 @@ pub(crate) struct DpnsMarketplaceSyncProgress {
     pub(crate) pending_departures: VecDeque<DpnsNameInfo>,
 }
 
+/// Names that left `identity_id` since the last completed ownership scan:
+/// every label the identity carries, plus every `Owned` marketplace row
+/// tracked for it, whose normalized label the scan did not see.
+///
+/// The label list is the primary trigger: a username fetch never drops a
+/// label from a wallet-owned identity (`apply_fetched_dpns_names` only
+/// merges there), so a departed name normally stays on the list until this
+/// sweep classifies it. The rows are belt-and-braces for a label list lost
+/// without its rows (a restore, or a host that persisted the rows but not
+/// the list): without them the sweep would never classify that departure,
+/// update its row, or refresh a seller's balance. Each normalized label is
+/// queued once, the label list first.
+fn departure_candidates(
+    identity_id: &Identifier,
+    previous_labels: &[DpnsNameInfo],
+    previous_rows: &BTreeMap<Identifier, DpnsNameStateEntry>,
+    seen_normalized_labels: &BTreeSet<String>,
+) -> VecDeque<DpnsNameInfo> {
+    let mut queued: BTreeSet<String> = BTreeSet::new();
+    let mut departures = VecDeque::new();
+    let from_labels = previous_labels
+        .iter()
+        .map(|name| (convert_to_homograph_safe_chars(&name.label), name.clone()));
+    let from_rows = previous_rows
+        .values()
+        .filter(|row| {
+            row.wallet_identity_id == *identity_id
+                && matches!(row.status, DpnsNameSaleStatus::Owned)
+        })
+        .map(|row| {
+            (
+                row.normalized_label.clone(),
+                DpnsNameInfo {
+                    label: row.label.clone(),
+                    acquired_at: row.transferred_at_ms.or(row.created_at_ms),
+                },
+            )
+        });
+    for (normalized, name) in from_labels.chain(from_rows) {
+        if seen_normalized_labels.contains(&normalized) || !queued.insert(normalized) {
+            continue;
+        }
+        departures.push_back(name);
+    }
+    departures
+}
+
 // ---------------------------------------------------------------------------
 // System contracts
 // ---------------------------------------------------------------------------
@@ -1639,15 +1686,12 @@ impl IdentityWallet {
 
                 if complete {
                     progress.cursor = None;
-                    progress.pending_departures = previous_labels
-                        .iter()
-                        .filter(|name| {
-                            !progress
-                                .seen_normalized_labels
-                                .contains(&convert_to_homograph_safe_chars(&name.label))
-                        })
-                        .cloned()
-                        .collect();
+                    progress.pending_departures = departure_candidates(
+                        &identity_id,
+                        &previous_labels,
+                        &previous_rows,
+                        &progress.seen_normalized_labels,
+                    );
                     progress.seen_normalized_labels.clear();
                 } else {
                     progress.cursor = next_cursor;
@@ -2371,6 +2415,9 @@ mod tests {
         /// mirror. Lets a test assert not merely what a sync summary
         /// CLAIMS but what actually reached the persistence boundary.
         stored_dpns: std::sync::Mutex<Vec<DpnsNameStateChangeSet>>,
+        /// The DPNS label list of every identity snapshot handed to
+        /// [`Self::store`], in order, keyed by identity id.
+        stored_identity_labels: std::sync::Mutex<Vec<(Identifier, Vec<String>)>>,
     }
 
     impl MirrorPersister {
@@ -2388,6 +2435,7 @@ mod tests {
                 lookups: std::sync::Mutex::new(Vec::new()),
                 fail: std::sync::Mutex::new(None),
                 stored_dpns: std::sync::Mutex::new(Vec::new()),
+                stored_identity_labels: std::sync::Mutex::new(Vec::new()),
             }
         }
 
@@ -2422,6 +2470,18 @@ mod tests {
                 .flat_map(|cs| cs.removed.iter().copied())
                 .collect()
         }
+
+        /// The DPNS label list of every stored snapshot of `identity_id`,
+        /// in store order.
+        fn stored_label_lists(&self, identity_id: &Identifier) -> Vec<Vec<String>> {
+            self.stored_identity_labels
+                .lock()
+                .expect("stored identity log")
+                .iter()
+                .filter(|(id, _)| id == identity_id)
+                .map(|(_, labels)| labels.clone())
+                .collect()
+        }
     }
 
     impl PlatformWalletPersistence for MirrorPersister {
@@ -2433,6 +2493,16 @@ mod tests {
             if let Some(cs) = changeset.dpns_name_states {
                 self.stored_dpns.lock().expect("stored log").push(cs);
             }
+            if let Some(cs) = changeset.identities {
+                let mut log = self
+                    .stored_identity_labels
+                    .lock()
+                    .expect("stored identity log");
+                for (id, entry) in cs.identities {
+                    let labels = entry.dpns_names.into_iter().map(|n| n.label).collect();
+                    log.push((id, labels));
+                }
+            }
             Ok(())
         }
 
@@ -3534,6 +3604,72 @@ mod tests {
         );
     }
 
+    /// A wallet identity whose label list lost a departed name (a restore
+    /// that brought back the marketplace rows but not the list — a username
+    /// fetch never prunes a wallet identity's labels) still has the name's
+    /// `Owned` row. The sweep must discover the departure from that row,
+    /// resolve it, and retire the row — otherwise the sale/transfer is never
+    /// classified and the stale row stays.
+    #[tokio::test]
+    async fn sync_pass_resolves_a_departure_whose_label_was_already_pruned() {
+        use dpp::identity::v0::IdentityV0;
+        use dpp::identity::Identity;
+
+        let document_id = Identifier::from([0xC1; 32]);
+        let identity_id = Identifier::from([0xC2; 32]);
+        let row = mirrored_row(document_id, identity_id);
+        let mirror = Arc::new(MirrorPersister::hydrated(vec![row.clone()]));
+        let wallet = mirror_backed_identity_wallet_with_sdk(
+            Arc::clone(&mirror),
+            sdk_for_departed_identity_sync(&identity_id, DEPARTED_LABEL).await,
+        );
+
+        // The identity does not carry the label (its list was lost); only
+        // the tracked marketplace row remembers the name.
+        {
+            let mut wm = wallet.wallet_manager.write().await;
+            let info = wm
+                .get_wallet_info_mut(&wallet.wallet_id)
+                .expect("wallet info");
+            info.identity_manager
+                .add_identity(
+                    Identity::V0(IdentityV0 {
+                        id: identity_id,
+                        public_keys: BTreeMap::new(),
+                        balance: 0,
+                        revision: 0,
+                    }),
+                    0,
+                    wallet.wallet_id,
+                    &wallet.persister,
+                )
+                .expect("add identity");
+            info.dpns_name_states.insert(document_id, row);
+        }
+        assert!(dpns_labels(&wallet, &identity_id).await.is_empty());
+
+        let summary = wallet
+            .sync_dpns_marketplace()
+            .await
+            .expect("sync pass must succeed");
+
+        assert_eq!(
+            summary.names_departed,
+            vec![DepartedDpnsName {
+                identity_id,
+                label: DEPARTED_LABEL.to_string(),
+                document_id: Some(document_id),
+                status: None,
+            }],
+            "the departure must be discovered from the tracked row"
+        );
+        assert_eq!(
+            mirror.stored_dpns_removals(),
+            vec![document_id],
+            "the departed row must be retired"
+        );
+    }
+
     /// A live `IdentityWallet` over a bare mock SDK (no expectations, so
     /// every network read fails) whose persister is `mirror`.
     fn mirror_backed_identity_wallet(mirror: Arc<MirrorPersister>) -> IdentityWallet {
@@ -4011,4 +4147,346 @@ mod tests {
              at the network: {error:?}"
         );
     }
+
+    // -----------------------------------------------------------------
+    // Username sync: the `DpnsFetch` each call site forwards
+    // -----------------------------------------------------------------
+
+    /// The identity a username sync test watches (out-of-wallet bucket, so
+    /// a complete fetch is allowed to drop names).
+    const WATCHED_IDENTITY: [u8; 32] = [0xD1; 32];
+
+    /// The label and document id of the `n`th fetched username; ids ascend
+    /// with `n`, as Platform returns them on the `records.identity` index.
+    fn fetched_username(n: u32) -> (String, Identifier) {
+        let mut id = [0xE0; 32];
+        id[28..].copy_from_slice(&n.to_be_bytes());
+        (format!("name-{n:04}"), Identifier::from(id))
+    }
+
+    /// A parseable DPNS `domain` document for the `n`th fetched username.
+    fn fetched_username_document(
+        identity_id: Identifier,
+        n: u32,
+    ) -> (Identifier, Option<Document>) {
+        let (label, id) = fetched_username(n);
+        let mut properties = BTreeMap::new();
+        properties.insert(
+            "normalizedLabel".to_string(),
+            Value::Text(convert_to_homograph_safe_chars(&label)),
+        );
+        properties.insert("label".to_string(), Value::Text(label));
+        properties.insert(
+            "normalizedParentDomainName".to_string(),
+            Value::Text(DPNS_PARENT_DOMAIN.to_string()),
+        );
+        properties.insert(
+            "records".to_string(),
+            Value::Map(vec![(
+                Value::Text("identity".to_string()),
+                Value::Identifier(identity_id.to_buffer()),
+            )]),
+        );
+        let document = Document::V0(dpp::document::DocumentV0 {
+            id,
+            owner_id: identity_id,
+            properties,
+            revision: Some(1),
+            ..Default::default()
+        });
+        (id, Some(document))
+    }
+
+    /// Serves the bundled DPNS contract (nothing else), the way a host's
+    /// context provider does, so the username query resolves its contract
+    /// without a network fetch or a dump directory.
+    struct DpnsContractProvider(Arc<DataContract>);
+
+    impl dash_sdk::platform::ContextProvider for DpnsContractProvider {
+        fn get_data_contract(
+            &self,
+            id: &Identifier,
+            _platform_version: &dpp::version::PlatformVersion,
+        ) -> Result<Option<Arc<DataContract>>, dash_sdk::error::ContextProviderError> {
+            Ok((*id == self.0.id()).then(|| Arc::clone(&self.0)))
+        }
+
+        fn get_token_configuration(
+            &self,
+            _token_id: &Identifier,
+        ) -> Result<
+            Option<dpp::data_contract::associated_token::token_configuration::TokenConfiguration>,
+            dash_sdk::error::ContextProviderError,
+        > {
+            Ok(None)
+        }
+
+        fn get_quorum_public_key(
+            &self,
+            _quorum_type: u32,
+            _quorum_hash: [u8; 32],
+            _core_chain_locked_height: u32,
+        ) -> Result<[u8; 48], dash_sdk::error::ContextProviderError> {
+            Err(dash_sdk::error::ContextProviderError::Config(
+                "no quorum keys in this test".to_string(),
+            ))
+        }
+
+        fn get_platform_activation_height(
+            &self,
+        ) -> Result<dpp::prelude::CoreBlockHeight, dash_sdk::error::ContextProviderError> {
+            Ok(1)
+        }
+    }
+
+    /// A mock SDK pinned to `version` answering the paged username query
+    /// for `identity_id` with pages of `page_sizes` consecutive usernames
+    /// (numbered from 0), each continuation keyed on the previous page's
+    /// last document id —
+    /// the exact requests `Sdk::get_all_dpns_usernames_by_identity` issues.
+    /// A request without an expectation fails the fetch.
+    async fn sdk_answering_username_pages(
+        version: &'static dpp::version::PlatformVersion,
+        identity_id: Identifier,
+        page_sizes: &[u32],
+    ) -> Arc<dash_sdk::Sdk> {
+        let contract = Arc::new(
+            dpp::system_data_contracts::load_system_data_contract(
+                dpp::data_contracts::SystemDataContract::DPNS,
+                dpp::version::PlatformVersion::latest(),
+            )
+            .expect("bundled DPNS contract"),
+        );
+        let mut sdk = dash_sdk::SdkBuilder::new_mock()
+            .with_version(version)
+            .with_context_provider(DpnsContractProvider(Arc::clone(&contract)))
+            .build()
+            .expect("mock sdk");
+        let mut start_after: Option<Identifier> = None;
+        let mut next = 0u32;
+        for &size in page_sizes {
+            let range = next..next + size;
+            next += size;
+            let query = DocumentQuery {
+                select: SelectProjection::documents(),
+                data_contract: Arc::clone(&contract),
+                document_type_name: DPNS_DOCUMENT_TYPE.to_string(),
+                where_clauses: vec![WhereClause {
+                    field: "records.identity".to_string(),
+                    operator: WhereOperator::Equal,
+                    value: Value::Identifier(identity_id.to_buffer()),
+                }],
+                time_range_clauses: vec![],
+                integer_range_clauses: vec![],
+                sub_queries: vec![],
+                group_by: vec![],
+                having: vec![],
+                order_by_clauses: vec![],
+                limit: DPNS_USERNAMES_PAGE_LIMIT,
+                offset: None,
+                start: start_after.map(|id| Start::StartAfter(id.to_vec())),
+            };
+            let page: dash_sdk::query_types::Documents = range
+                .clone()
+                .map(|n| fetched_username_document(identity_id, n))
+                .collect();
+            start_after = range.last().map(|n| fetched_username(n).1);
+            sdk.mock()
+                .expect_fetch_many::<Identifier, Document, _, dash_sdk::query_types::Documents>(
+                    query,
+                    Some(page),
+                )
+                .await
+                .expect("username page expectation");
+        }
+        Arc::new(sdk)
+    }
+
+    /// A wallet over `sdk` watching [`WATCHED_IDENTITY`], which already
+    /// carries `known` labels (a name that left it, plus the first fetched
+    /// one). Returns the wallet, its mirror and the identity id.
+    async fn wallet_watching_identity_with(
+        sdk: Arc<dash_sdk::Sdk>,
+        known: &[&str],
+    ) -> (IdentityWallet, Arc<MirrorPersister>, Identifier) {
+        use dpp::identity::v0::IdentityV0;
+        use dpp::identity::Identity;
+
+        let identity_id = Identifier::from(WATCHED_IDENTITY);
+        let mirror = Arc::new(MirrorPersister::hydrated(Vec::new()));
+        let wallet = mirror_backed_identity_wallet_with_sdk(Arc::clone(&mirror), sdk);
+        {
+            let mut wm = wallet.wallet_manager.write().await;
+            let info = wm
+                .get_wallet_info_mut(&wallet.wallet_id)
+                .expect("wallet info");
+            info.identity_manager
+                .add_out_of_wallet_identity(
+                    Identity::V0(IdentityV0 {
+                        id: identity_id,
+                        public_keys: BTreeMap::new(),
+                        balance: 0,
+                        revision: 0,
+                    }),
+                    &wallet.persister,
+                )
+                .expect("watch identity");
+            info.identity_manager
+                .managed_identity_mut(&identity_id)
+                .expect("watched identity")
+                .set_dpns_names(
+                    known
+                        .iter()
+                        .map(|label| DpnsNameInfo {
+                            label: label.to_string(),
+                            acquired_at: Some(1),
+                        })
+                        .collect(),
+                    &wallet.persister,
+                );
+        }
+        (wallet, mirror, identity_id)
+    }
+
+    async fn watched_labels(wallet: &IdentityWallet, identity_id: &Identifier) -> Vec<String> {
+        let wm = wallet.wallet_manager.read().await;
+        wm.get_wallet_info(&wallet.wallet_id)
+            .expect("wallet info")
+            .identity_manager
+            .managed_identity(identity_id)
+            .expect("watched identity")
+            .dpns_names
+            .iter()
+            .map(|name| name.label.clone())
+            .collect()
+    }
+
+    fn username_labels(range: std::ops::Range<u32>) -> Vec<String> {
+        range.map(|n| fetched_username(n).0).collect()
+    }
+
+    /// A full page followed by a short one is the identity's whole owned
+    /// set: `sync_dpns_names` must forward `DpnsFetch::Complete`, so the
+    /// watched identity's list is replaced (the departed name drops out) in
+    /// a single store.
+    #[tokio::test]
+    async fn sync_dpns_names_replaces_a_watched_list_after_a_short_page() {
+        let limit = DPNS_USERNAMES_PAGE_LIMIT;
+        let identity_id = Identifier::from(WATCHED_IDENTITY);
+        let sdk = sdk_answering_username_pages(
+            dpp::version::PlatformVersion::latest(),
+            identity_id,
+            &[limit, 1],
+        )
+        .await;
+        let first = fetched_username(0).0;
+        let (wallet, mirror, identity_id) =
+            wallet_watching_identity_with(sdk, &["departed", &first]).await;
+        let stores_before = mirror.stored_label_lists(&identity_id).len();
+
+        let added = wallet
+            .sync_dpns_names(&identity_id)
+            .await
+            .expect("both pages answer");
+
+        let expected = username_labels(0..limit + 1);
+        assert_eq!(added, limit, "every fetched name but the known first one");
+        assert_eq!(watched_labels(&wallet, &identity_id).await, expected);
+        let stores = mirror.stored_label_lists(&identity_id);
+        assert_eq!(
+            &stores[stores_before..],
+            [expected],
+            "one snapshot for the whole fetch"
+        );
+    }
+
+    /// Reaching the page budget with every page full proves nothing about
+    /// what lies beyond: `sync_dpns_names` must forward `DpnsFetch::Partial`,
+    /// so the watched identity's list only grows.
+    #[tokio::test]
+    async fn sync_dpns_names_only_grows_a_watched_list_when_the_page_budget_runs_out() {
+        let limit = DPNS_USERNAMES_PAGE_LIMIT;
+        let max_pages = DPNS_USERNAMES_MAX_PAGES as u32;
+        let identity_id = Identifier::from(WATCHED_IDENTITY);
+        let sdk = sdk_answering_username_pages(
+            dpp::version::PlatformVersion::latest(),
+            identity_id,
+            &vec![limit; max_pages as usize],
+        )
+        .await;
+        let first = fetched_username(0).0;
+        let (wallet, mirror, identity_id) =
+            wallet_watching_identity_with(sdk, &["kept", &first]).await;
+        let stores_before = mirror.stored_label_lists(&identity_id).len();
+
+        let added = wallet
+            .sync_dpns_names(&identity_id)
+            .await
+            .expect("every page within the budget answers");
+
+        let mut expected = vec!["kept".to_string()];
+        expected.extend(username_labels(0..max_pages * limit));
+        assert_eq!(added, max_pages * limit - 1);
+        assert_eq!(
+            watched_labels(&wallet, &identity_id).await,
+            expected,
+            "a lower bound must not drop the name it did not see"
+        );
+        assert_eq!(
+            mirror.stored_label_lists(&identity_id).len(),
+            stores_before + 1
+        );
+    }
+
+    /// On a platform version whose query lowering cannot continue a page
+    /// (protocol version <= 13), a full first page is all that is read, and
+    /// it is only a lower bound: the watched list only grows. Only the first
+    /// page has an expectation, so a continuation would fail the sync.
+    #[tokio::test]
+    async fn sync_dpns_names_only_grows_a_watched_list_after_a_full_page_on_the_v0_lowering() {
+        let limit = DPNS_USERNAMES_PAGE_LIMIT;
+        let v13 = dpp::version::PlatformVersion::get(13).expect("protocol version 13");
+        let identity_id = Identifier::from(WATCHED_IDENTITY);
+        let sdk = sdk_answering_username_pages(v13, identity_id, &[limit]).await;
+        let (wallet, _mirror, identity_id) = wallet_watching_identity_with(sdk, &["kept"]).await;
+
+        let added = wallet
+            .sync_dpns_names(&identity_id)
+            .await
+            .expect("only the first page is requested");
+
+        let mut expected = vec!["kept".to_string()];
+        expected.extend(username_labels(0..limit));
+        assert_eq!(added, limit);
+        assert_eq!(watched_labels(&wallet, &identity_id).await, expected);
+    }
+
+    /// `refresh_dpns_names` forwards the same `DpnsFetch`: a complete fetch
+    /// replaces a watched identity's list in one store.
+    #[tokio::test]
+    async fn refresh_dpns_names_replaces_a_watched_list_after_a_complete_fetch() {
+        let identity_id = Identifier::from(WATCHED_IDENTITY);
+        let sdk = sdk_answering_username_pages(
+            dpp::version::PlatformVersion::latest(),
+            identity_id,
+            &[2],
+        )
+        .await;
+        let first = fetched_username(0).0;
+        let (wallet, mirror, identity_id) =
+            wallet_watching_identity_with(sdk, &["departed", &first]).await;
+        let stores_before = mirror.stored_label_lists(&identity_id).len();
+
+        wallet
+            .refresh_dpns_names()
+            .await
+            .expect("refresh must succeed");
+
+        let expected = username_labels(0..2);
+        assert_eq!(watched_labels(&wallet, &identity_id).await, expected);
+        assert_eq!(
+            &mirror.stored_label_lists(&identity_id)[stores_before..],
+            [expected]
+        );
+    }
 }
diff --git a/packages/rs-platform-wallet/src/wallet/identity/network/loading.rs b/packages/rs-platform-wallet/src/wallet/identity/network/loading.rs
index ecc7897ae4b..3133c1e2bea 100644
--- a/packages/rs-platform-wallet/src/wallet/identity/network/loading.rs
+++ b/packages/rs-platform-wallet/src/wallet/identity/network/loading.rs
@@ -166,7 +166,6 @@ impl IdentityWallet {
         identity_index: u32,
         source: LoadKeyHashSource<'_>,
     ) -> Result<Option<Identity>, PlatformWalletError> {
-        use crate::wallet::identity::state::managed_identity::key_storage::DpnsNameInfo;
         use crate::wallet::identity::state::managed_identity::key_storage::IdentityStatus;
         use dash_sdk::platform::types::identity::PublicKeyHash;
         use dash_sdk::platform::Fetch;
@@ -278,12 +277,8 @@ impl IdentityWallet {
         }
 
         // Query DPNS names for the discovered identity.
-        match self
-            .sdk
-            .get_dpns_usernames_by_identity(identity_id, None)
-            .await
-        {
-            Ok(usernames) => {
+        match self.fetch_owned_dpns_names(identity_id, None).await {
+            Ok((names, fetch)) => {
                 let mut wm = self.wallet_manager.write().await;
                 let info = wm.get_wallet_info_mut(&self.wallet_id).ok_or_else(|| {
                     crate::error::PlatformWalletError::WalletNotFound(
@@ -291,15 +286,7 @@ impl IdentityWallet {
                     )
                 })?;
                 if let Some(managed) = info.identity_manager.managed_identity_mut(&identity_id) {
-                    for username in usernames {
-                        managed.add_dpns_name(
-                            DpnsNameInfo {
-                                label: username.label,
-                                acquired_at: None,
-                            },
-                            &self.persister,
-                        );
-                    }
+                    managed.apply_fetched_dpns_names(names, fetch, &self.persister);
                 }
             }
             Err(e) => {
@@ -408,11 +395,13 @@ impl IdentityWallet {
     /// Refresh DPNS names for all identities in the manager.
     ///
     /// Iterates every identity in the [`IdentityManager`], queries Platform
-    /// for its current DPNS usernames, and replaces the stored
-    /// `dpns_names` list with the fresh results.
+    /// for its current DPNS usernames, and reconciles the stored
+    /// `dpns_names` list with them, persisting at most one snapshot per
+    /// identity: for an identity the wallet only watches, a complete result
+    /// replaces the list (departed names drop out); wallet-owned identities
+    /// and a possibly truncated result only add — see
+    /// `ManagedIdentity::apply_fetched_dpns_names`.
     pub async fn refresh_dpns_names(&self) -> Result<(), PlatformWalletError> {
-        use crate::wallet::identity::state::managed_identity::key_storage::DpnsNameInfo;
-
         // Collect identity IDs so we don't hold the lock during network calls.
         let identity_ids: Vec<Identifier> = {
             let wm = self.wallet_manager.read().await;
@@ -429,12 +418,8 @@ impl IdentityWallet {
         };
 
         for identity_id in identity_ids {
-            match self
-                .sdk
-                .get_dpns_usernames_by_identity(identity_id, None)
-                .await
-            {
-                Ok(usernames) => {
+            match self.fetch_owned_dpns_names(identity_id, None).await {
+                Ok((names, fetch)) => {
                     let mut wm = self.wallet_manager.write().await;
                     let info = wm.get_wallet_info_mut(&self.wallet_id).ok_or_else(|| {
                         crate::error::PlatformWalletError::WalletNotFound(
@@ -443,13 +428,7 @@ impl IdentityWallet {
                     })?;
                     if let Some(managed) = info.identity_manager.managed_identity_mut(&identity_id)
                     {
-                        managed.dpns_names = usernames
-                            .into_iter()
-                            .map(|u| DpnsNameInfo {
-                                label: u.label,
-                                acquired_at: None,
-                            })
-                            .collect();
+                        managed.apply_fetched_dpns_names(names, fetch, &self.persister);
                     }
                 }
                 Err(e) => {
diff --git a/packages/rs-platform-wallet/src/wallet/identity/network/mod.rs b/packages/rs-platform-wallet/src/wallet/identity/network/mod.rs
index 4bd019c8392..7186dddca96 100644
--- a/packages/rs-platform-wallet/src/wallet/identity/network/mod.rs
+++ b/packages/rs-platform-wallet/src/wallet/identity/network/mod.rs
@@ -19,6 +19,18 @@
 //! broadcaster used by DashPay `send_payment`; it defaults to
 //! `SpvBroadcaster` so most call sites don't need to name it.
 
+use dash_sdk::drive::config::DEFAULT_MAX_QUERY_LIMIT;
+
+/// Page size for the per-identity DPNS username query: Platform's maximum
+/// query limit.
+pub(crate) const DPNS_USERNAMES_PAGE_LIMIT: u32 = DEFAULT_MAX_QUERY_LIMIT as u32;
+
+/// Pages read per identity before giving up on completeness. Paging ends as
+/// soon as a page comes back short (the complete owned set); reaching this
+/// bound means the result is only a lower bound and is merged, never used to
+/// drop names — see `ManagedIdentity::apply_fetched_dpns_names`.
+pub(crate) const DPNS_USERNAMES_MAX_PAGES: usize = 10;
+
 // Core handle + identity-lifecycle operations.
 mod balance;
 mod contract;
diff --git a/packages/rs-platform-wallet/src/wallet/identity/state/managed_identity/identity_ops.rs b/packages/rs-platform-wallet/src/wallet/identity/state/managed_identity/identity_ops.rs
index 1a4a839651a..0e380b080e6 100644
--- a/packages/rs-platform-wallet/src/wallet/identity/state/managed_identity/identity_ops.rs
+++ b/packages/rs-platform-wallet/src/wallet/identity/state/managed_identity/identity_ops.rs
@@ -1,6 +1,6 @@
 //! Core identity operations for ManagedIdentity
 
-use super::key_storage::{DpnsNameInfo, IdentityStatus};
+use super::key_storage::{DpnsFetch, DpnsNameInfo, IdentityStatus};
 use super::ManagedIdentity;
 use crate::changeset::{IdentityChangeSet, IdentityEntry, IdentityKeyEntry, IdentityKeysChangeSet};
 use crate::wallet::persister::WalletPersister;
@@ -242,6 +242,80 @@ impl ManagedIdentity {
         }
     }
 
+    /// Merge fetched DPNS names into the list, skipping labels already
+    /// present, and persist the result as one snapshot. Returns how many
+    /// names were added; nothing is persisted when that is zero.
+    ///
+    /// Use this for a batch instead of calling [`Self::add_dpns_name`] per
+    /// name: every persisted snapshot is read downstream as the identity's
+    /// complete owned set, so a snapshot per name makes the first one look
+    /// like the others were lost. Add-only — DPNS queries are capped by a
+    /// limit, so a label missing from a fetch is not proof it left.
+    pub fn merge_dpns_names(
+        &mut self,
+        names: impl IntoIterator<Item = DpnsNameInfo>,
+        persister: &WalletPersister,
+    ) -> u32 {
+        let mut merged = self.dpns_names.clone();
+        let mut added = 0u32;
+        for name in names {
+            if merged.iter().any(|existing| existing.label == name.label) {
+                continue;
+            }
+            merged.push(name);
+            added += 1;
+        }
+        if added > 0 {
+            self.set_dpns_names(merged, persister);
+        }
+        added
+    }
+
+    /// Apply one DPNS username fetch for this identity; returns how many
+    /// labels are new.
+    ///
+    /// On a [`DpnsFetch::Complete`] fetch (the identity's whole owned set)
+    /// of an identity the wallet only watches (no `wallet_id`) the list is
+    /// replaced with it — departed names drop out, an empty result clears
+    /// the list — while labels already known keep their `acquired_at`, and
+    /// nothing is persisted when the list is unchanged.
+    ///
+    /// A wallet-owned identity only ever merges new labels
+    /// ([`Self::merge_dpns_names`]): the DPNS marketplace sweep discovers its
+    /// departures from this list and classifies them (sale, transfer,
+    /// deletion), so a username fetch must not remove a label first. A
+    /// [`DpnsFetch::Partial`] fetch cannot prove that a missing label left,
+    /// so it only merges too. Either way one snapshot at most.
+    pub fn apply_fetched_dpns_names(
+        &mut self,
+        names: Vec<DpnsNameInfo>,
+        fetch: DpnsFetch,
+        persister: &WalletPersister,
+    ) -> u32 {
+        if fetch == DpnsFetch::Partial || self.wallet_id.is_some() {
+            return self.merge_dpns_names(names, persister);
+        }
+        // Known labels still owned, in their existing order, then new ones.
+        let mut next: Vec<DpnsNameInfo> = self
+            .dpns_names
+            .iter()
+            .filter(|known| names.iter().any(|name| name.label == known.label))
+            .cloned()
+            .collect();
+        let mut added = 0u32;
+        for name in names {
+            if next.iter().any(|existing| existing.label == name.label) {
+                continue;
+            }
+            next.push(name);
+            added += 1;
+        }
+        if next != self.dpns_names {
+            self.set_dpns_names(next, persister);
+        }
+        added
+    }
+
     /// Replace the DPNS-name list wholesale.
     ///
     /// Use this when a sync round (or a confirmed sale/transfer) has the
@@ -624,6 +698,201 @@ mod tests {
         })
     }
 
+    /// A batch of fetched names lands as one snapshot carrying every name,
+    /// skips labels already held, and persists nothing when all are known.
+    /// One snapshot per name made the first read as the complete owned set.
+    #[test]
+    fn merge_dpns_names_persists_one_complete_snapshot() {
+        let identity = Identity::V0(IdentityV0 {
+            id: Identifier::from([1u8; 32]),
+            public_keys: BTreeMap::new(),
+            balance: 0,
+            revision: 0,
+        });
+        let mut managed = ManagedIdentity::new(identity, 0);
+        let persister = std::sync::Arc::new(CapturingPersister::default());
+        let p = WalletPersister::new([0xAB; 32], std::sync::Arc::clone(&persister) as _);
+        let name = |label: &str| DpnsNameInfo {
+            label: label.to_string(),
+            acquired_at: None,
+        };
+
+        managed.add_dpns_name(name("alice"), &p);
+        let added = managed.merge_dpns_names([name("alice"), name("bob"), name("carol")], &p);
+
+        assert_eq!(added, 2);
+        let stores = persister.stores.lock().unwrap();
+        assert_eq!(
+            stores.len(),
+            2,
+            "one store for add_dpns_name, one for the merge"
+        );
+        let labels: Vec<_> = stores[1]
+            .identities
+            .as_ref()
+            .expect("identity snapshot")
+            .identities
+            .values()
+            .next()
+            .expect("one identity")
+            .dpns_names
+            .iter()
+            .map(|n| n.label.as_str())
+            .collect();
+        assert_eq!(labels, ["alice", "bob", "carol"]);
+        drop(stores);
+
+        assert_eq!(managed.merge_dpns_names([name("bob")], &p), 0);
+        assert_eq!(
+            persister.stores.lock().unwrap().len(),
+            2,
+            "nothing new, nothing stored"
+        );
+    }
+
+    fn dpns_test_identity() -> (
+        ManagedIdentity,
+        std::sync::Arc<CapturingPersister>,
+        WalletPersister,
+    ) {
+        let identity = Identity::V0(IdentityV0 {
+            id: Identifier::from([1u8; 32]),
+            public_keys: BTreeMap::new(),
+            balance: 0,
+            revision: 0,
+        });
+        let persister = std::sync::Arc::new(CapturingPersister::default());
+        let p = WalletPersister::new([0xAB; 32], std::sync::Arc::clone(&persister) as _);
+        (ManagedIdentity::new(identity, 0), persister, p)
+    }
+
+    fn dpns_name(label: &str, acquired_at: Option<u64>) -> DpnsNameInfo {
+        DpnsNameInfo {
+            label: label.to_string(),
+            acquired_at,
+        }
+    }
+
+    fn dpns_store_count(persister: &CapturingPersister) -> usize {
+        persister.stores.lock().unwrap().len()
+    }
+
+    /// A complete fetch is the owned set: a departed label drops out, a known
+    /// label keeps its timestamp, a new one is added — in one store.
+    #[test]
+    fn complete_dpns_fetch_replaces_the_list_in_one_store() {
+        let (mut managed, persister, p) = dpns_test_identity();
+        managed.set_dpns_names(
+            vec![dpns_name("alice", Some(10)), dpns_name("bob", Some(20))],
+            &p,
+        );
+        let before = dpns_store_count(&persister);
+
+        let added = managed.apply_fetched_dpns_names(
+            vec![dpns_name("bob", Some(99)), dpns_name("carol", Some(99))],
+            DpnsFetch::Complete,
+            &p,
+        );
+
+        assert_eq!(added, 1);
+        assert_eq!(
+            managed.dpns_names,
+            vec![dpns_name("bob", Some(20)), dpns_name("carol", Some(99))]
+        );
+        assert_eq!(dpns_store_count(&persister), before + 1);
+    }
+
+    /// An empty complete fetch means the identity owns no names any more.
+    #[test]
+    fn empty_complete_dpns_fetch_clears_the_list() {
+        let (mut managed, persister, p) = dpns_test_identity();
+        managed.set_dpns_names(vec![dpns_name("alice", Some(10))], &p);
+        let before = dpns_store_count(&persister);
+
+        assert_eq!(
+            managed.apply_fetched_dpns_names(Vec::new(), DpnsFetch::Complete, &p),
+            0
+        );
+
+        assert!(managed.dpns_names.is_empty());
+        assert_eq!(dpns_store_count(&persister), before + 1);
+    }
+
+    /// A full page may be truncated: a label missing from it stays.
+    #[test]
+    fn partial_dpns_fetch_only_adds() {
+        let (mut managed, _persister, p) = dpns_test_identity();
+        managed.set_dpns_names(vec![dpns_name("alice", Some(10))], &p);
+
+        let added =
+            managed.apply_fetched_dpns_names(vec![dpns_name("bob", None)], DpnsFetch::Partial, &p);
+
+        assert_eq!(added, 1);
+        assert_eq!(
+            managed.dpns_names,
+            vec![dpns_name("alice", Some(10)), dpns_name("bob", None)]
+        );
+    }
+
+    /// A wallet-owned identity never loses a label to a username fetch, even
+    /// a complete one (including an empty complete fetch): it only gains the
+    /// new labels. The marketplace sweep discovers its departures from this
+    /// list, so pruning here would erase the only trigger.
+    #[test]
+    fn complete_dpns_fetch_on_a_wallet_identity_only_adds() {
+        let (mut managed, _persister, p) = dpns_test_identity();
+        managed.wallet_id = Some([0xAB; 32]);
+        managed.set_dpns_names(
+            vec![dpns_name("alice", Some(10)), dpns_name("bob", Some(20))],
+            &p,
+        );
+
+        let added = managed.apply_fetched_dpns_names(
+            vec![dpns_name("carol", None)],
+            DpnsFetch::Complete,
+            &p,
+        );
+        assert_eq!(added, 1);
+        let labels: Vec<&str> = managed
+            .dpns_names
+            .iter()
+            .map(|n| n.label.as_str())
+            .collect();
+        assert_eq!(labels, ["alice", "bob", "carol"]);
+
+        assert_eq!(
+            managed.apply_fetched_dpns_names(Vec::new(), DpnsFetch::Complete, &p),
+            0
+        );
+        assert_eq!(
+            managed.dpns_names.len(),
+            3,
+            "an empty complete fetch prunes nothing"
+        );
+    }
+
+    /// Re-fetching the same complete set (even in another order, with fresh
+    /// timestamps) changes nothing and stores nothing.
+    #[test]
+    fn unchanged_complete_dpns_fetch_stores_nothing() {
+        let (mut managed, persister, p) = dpns_test_identity();
+        managed.set_dpns_names(
+            vec![dpns_name("alice", Some(10)), dpns_name("bob", Some(20))],
+            &p,
+        );
+        let before = dpns_store_count(&persister);
+
+        let added = managed.apply_fetched_dpns_names(
+            vec![dpns_name("bob", Some(99)), dpns_name("alice", Some(99))],
+            DpnsFetch::Complete,
+            &p,
+        );
+
+        assert_eq!(added, 0);
+        assert_eq!(dpns_store_count(&persister), before);
+        assert_eq!(managed.dpns_names[0], dpns_name("alice", Some(10)));
+    }
+
     /// `add_keys` records each key's breadcrumb (or `None` for watch-only)
     /// in one batched changeset and lands every key in the DPP identity.
     /// Pins the materialization side of the imported-identity-signing fix.
diff --git a/packages/rs-platform-wallet/src/wallet/identity/state/managed_identity/mod.rs b/packages/rs-platform-wallet/src/wallet/identity/state/managed_identity/mod.rs
index 6038779d572..a7c3eafaaff 100644
--- a/packages/rs-platform-wallet/src/wallet/identity/state/managed_identity/mod.rs
+++ b/packages/rs-platform-wallet/src/wallet/identity/state/managed_identity/mod.rs
@@ -17,7 +17,7 @@ pub use dashpay::DashPayState;
 // `state::managed_identity::*` path.
 pub use crate::wallet::identity::types::block_time::{self, BlockTime};
 pub use crate::wallet::identity::types::key_storage::{
-    self, DpnsNameInfo, IdentityStatus, KeyStorage, PrivateKeyData,
+    self, DpnsFetch, DpnsNameInfo, IdentityStatus, KeyStorage, PrivateKeyData,
 };
 
 use dpp::identity::Identity;
diff --git a/packages/rs-platform-wallet/src/wallet/identity/types/key_storage.rs b/packages/rs-platform-wallet/src/wallet/identity/types/key_storage.rs
index 634a6459e3f..f2153841956 100644
--- a/packages/rs-platform-wallet/src/wallet/identity/types/key_storage.rs
+++ b/packages/rs-platform-wallet/src/wallet/identity/types/key_storage.rs
@@ -58,6 +58,17 @@ pub struct DpnsNameInfo {
     pub acquired_at: Option<u64>,
 }
 
+/// How much of an identity's owned DPNS names one username fetch returned.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum DpnsFetch {
+    /// The identity's whole owned set: paging reached a short page, so a
+    /// label missing from the fetch has left the identity.
+    Complete,
+    /// A lower bound: the page bound was reached, or the platform version
+    /// cannot continue a page. A label missing from the fetch proves nothing.
+    Partial,
+}
+
 /// Private key storage mapping KeyID to public key metadata + private key data.
 ///
 /// Lives only in transient places — the `IdentityKeysChangeSet` apply
diff --git a/packages/rs-platform-wallet/src/wallet/identity/types/mod.rs b/packages/rs-platform-wallet/src/wallet/identity/types/mod.rs
index 288f88a021c..c6623c5a8b2 100644
--- a/packages/rs-platform-wallet/src/wallet/identity/types/mod.rs
+++ b/packages/rs-platform-wallet/src/wallet/identity/types/mod.rs
@@ -14,4 +14,4 @@ pub use dashpay::{
     ContactProfileEntry, ContactRequest, DashPayProfile, DashpayAddressMatch, EstablishedContact,
     PaymentDirection, PaymentEntry, PaymentStatus, ProfileUpdate,
 };
-pub use key_storage::{DpnsNameInfo, IdentityStatus, KeyStorage, PrivateKeyData};
+pub use key_storage::{DpnsFetch, DpnsNameInfo, IdentityStatus, KeyStorage, PrivateKeyData};
diff --git a/packages/rs-sdk/src/platform/dpns_usernames/queries.rs b/packages/rs-sdk/src/platform/dpns_usernames/queries.rs
index d3f22d40e25..80c67ea5cdd 100644
--- a/packages/rs-sdk/src/platform/dpns_usernames/queries.rs
+++ b/packages/rs-sdk/src/platform/dpns_usernames/queries.rs
@@ -1,10 +1,14 @@
 use crate::platform::documents::document_query::DocumentQuery;
 use crate::platform::{Document, FetchMany};
 use crate::{Error, Sdk};
+use dapi_grpc::platform::v0::get_documents_request::get_documents_request_v0::Start;
+use dpp::data_contract::DataContract;
 use dpp::document::DocumentV0Getters;
 use dpp::platform_value::Value;
 use dpp::prelude::Identifier;
-use drive::query::{OrderClause, WhereClause, WhereOperator};
+use dpp::version::PlatformVersion;
+use drive::query::{OrderClause, SelectProjection, WhereClause, WhereOperator};
+use std::sync::Arc;
 
 use super::convert_to_homograph_safe_chars;
 
@@ -80,6 +84,69 @@ impl Sdk {
         Ok(usernames)
     }
 
+    /// Get every DPNS username associated with `identity_id`, paging through
+    /// the `records.identity` index `page_limit` documents at a time.
+    ///
+    /// Returns the usernames and whether the set is known to be complete:
+    /// `true` once a page comes back shorter than `page_limit`, `false` when
+    /// the end was not provably reached (the caller must then treat the
+    /// result as a lower bound, not the owned set). Documents that do not
+    /// parse as DPNS domains are skipped, as in
+    /// [`Self::get_dpns_usernames_by_identity`].
+    ///
+    /// # When `true` can be trusted
+    ///
+    /// Continuing a page relies on `startAfter` inside the `records.identity`
+    /// equality bucket, which only the query lowering introduced with
+    /// protocol version 14 (`non_primary_key_path_query: 1`) serves. The
+    /// earlier lowering drops the rest of the bucket after any cursor, so a
+    /// second page would come back empty and read as the end. On such a
+    /// platform version (the one the SDK runs at once the first page has
+    /// answered) the loop reads only the first page: `true` then means the
+    /// identity has fewer than `page_limit` names, and a full first page
+    /// returns `false`. Otherwise `false` means `max_pages` full pages were
+    /// read without reaching the end.
+    pub async fn get_all_dpns_usernames_by_identity(
+        &self,
+        identity_id: Identifier,
+        page_limit: u32,
+        max_pages: usize,
+    ) -> Result<(Vec<DpnsUsername>, bool), Error> {
+        let dpns_contract = self.fetch_dpns_contract().await?;
+        let mut usernames = Vec::new();
+        let mut start_after: Option<Identifier> = None;
+        for _ in 0..max_pages {
+            let query = identity_domains_page_query(
+                dpns_contract.clone(),
+                identity_id,
+                page_limit,
+                start_after,
+            );
+            let documents = Document::fetch_many(self, query).await?;
+            // Termination and the cursor follow the RAW page (every returned
+            // entry, parseable or not), never the converted usernames: a
+            // skipped document still counts toward a full page and still
+            // moves the cursor past it.
+            let page_len = documents.len();
+            let last_id = documents.keys().last().copied();
+            // Read after the fetch: the response has already moved an
+            // auto-detecting SDK to the network's protocol version.
+            let cursor_continues = start_after_continues_an_equality_bucket(self.version());
+            usernames.extend(
+                documents
+                    .into_values()
+                    .flatten()
+                    .filter_map(Self::document_to_dpns_username),
+            );
+            match next_dpns_page(page_len, page_limit, last_id, cursor_continues) {
+                DpnsPageStep::Complete => return Ok((usernames, true)),
+                DpnsPageStep::Continue(cursor) => start_after = Some(cursor),
+                DpnsPageStep::Unknown => return Ok((usernames, false)),
+            }
+        }
+        Ok((usernames, false))
+    }
+
     /// Check if a DPNS username is available
     ///
     /// # Arguments
@@ -201,3 +268,422 @@ impl Sdk {
         })
     }
 }
+
+/// One page of `domain` documents whose `records.identity` is `identity_id`,
+/// resuming after `start_after` when set.
+fn identity_domains_page_query(
+    dpns_contract: Arc<DataContract>,
+    identity_id: Identifier,
+    page_limit: u32,
+    start_after: Option<Identifier>,
+) -> DocumentQuery {
+    DocumentQuery {
+        select: SelectProjection::documents(),
+        data_contract: dpns_contract,
+        document_type_name: "domain".to_string(),
+        where_clauses: vec![WhereClause {
+            field: "records.identity".to_string(),
+            operator: WhereOperator::Equal,
+            value: Value::Identifier(identity_id.to_buffer()),
+        }],
+        time_range_clauses: vec![],
+        integer_range_clauses: vec![],
+        sub_queries: vec![],
+        group_by: vec![],
+        having: vec![],
+        order_by_clauses: vec![],
+        limit: page_limit,
+        offset: None,
+        start: start_after.map(|id| Start::StartAfter(id.to_vec())),
+    }
+}
+
+/// What to do after one page of a paged DPNS username query.
+#[derive(Debug, PartialEq, Eq)]
+enum DpnsPageStep {
+    /// A short page: the whole set has been read.
+    Complete,
+    /// A full page: continue after this document id.
+    Continue(Identifier),
+    /// A full page that cannot be continued (no cursor, or a platform
+    /// version whose lowering drops the rest of the bucket after one):
+    /// completeness cannot be established.
+    Unknown,
+}
+
+fn next_dpns_page(
+    page_len: usize,
+    page_limit: u32,
+    last_id: Option<Identifier>,
+    cursor_continues: bool,
+) -> DpnsPageStep {
+    if page_len < page_limit as usize {
+        DpnsPageStep::Complete
+    } else if let (Some(id), true) = (last_id, cursor_continues) {
+        DpnsPageStep::Continue(id)
+    } else {
+        DpnsPageStep::Unknown
+    }
+}
+
+/// Whether `startAfter` a document inside an equality bucket (no `orderBy`)
+/// returns the rest of that bucket. The v0 non-primary-key lowering
+/// (protocol versions <= 13) hands the exclusive cursor to the equality
+/// key itself and so drops the whole bucket; the v1 lowering continues on
+/// the document-id level. Pinned in Drive by the
+/// `dpns_identity_records_paging` tests in `rs-drive/tests/query_tests.rs`.
+fn start_after_continues_an_equality_bucket(platform_version: &PlatformVersion) -> bool {
+    platform_version
+        .drive
+        .methods
+        .document
+        .query
+        .non_primary_key_path_query
+        != 0
+}
+
+#[cfg(test)]
+mod paging_tests {
+    use super::*;
+
+    #[test]
+    fn a_short_page_completes_the_set() {
+        for cursor_continues in [true, false] {
+            assert_eq!(
+                next_dpns_page(0, 100, None, cursor_continues),
+                DpnsPageStep::Complete
+            );
+            assert_eq!(
+                next_dpns_page(99, 100, Some(Identifier::from([1u8; 32])), cursor_continues),
+                DpnsPageStep::Complete
+            );
+        }
+    }
+
+    #[test]
+    fn a_full_page_continues_after_its_last_document() {
+        let last = Identifier::from([7u8; 32]);
+        assert_eq!(
+            next_dpns_page(100, 100, Some(last), true),
+            DpnsPageStep::Continue(last)
+        );
+    }
+
+    #[test]
+    fn a_full_page_without_a_cursor_is_not_complete() {
+        assert_eq!(next_dpns_page(100, 100, None, true), DpnsPageStep::Unknown);
+    }
+
+    #[test]
+    fn a_full_page_on_a_lowering_that_drops_the_bucket_is_not_complete() {
+        assert_eq!(
+            next_dpns_page(100, 100, Some(Identifier::from([7u8; 32])), false),
+            DpnsPageStep::Unknown
+        );
+    }
+
+    #[test]
+    fn only_the_v1_lowering_continues_an_equality_bucket() {
+        for protocol_version in 1..=13 {
+            let version = PlatformVersion::get(protocol_version).expect("released version");
+            assert!(
+                !start_after_continues_an_equality_bucket(version),
+                "protocol version {protocol_version} uses the v0 lowering"
+            );
+        }
+        assert!(start_after_continues_an_equality_bucket(
+            PlatformVersion::latest()
+        ));
+    }
+}
+
+/// The real paging loop against a network-free mock SDK. Every expectation
+/// is keyed by the exact encoded request, so a wrong `StartAfter` cursor, a
+/// wrong limit or an extra page fails the fetch instead of passing silently.
+#[cfg(all(test, feature = "mocks"))]
+mod paging_loop_tests {
+    use super::*;
+    use crate::platform::dpns_usernames::convert_to_homograph_safe_chars;
+    use crate::query_types::Documents;
+    use crate::SdkBuilder;
+    use dpp::data_contract::accessors::v0::{DataContractV0Getters, DataContractV0Setters};
+    use dpp::data_contract::DataContract;
+    use dpp::document::DocumentV0;
+    use dpp::version::PlatformVersion;
+    use std::collections::BTreeMap;
+    use std::sync::Arc;
+
+    const PAGE: u32 = 2;
+
+    fn identity() -> Identifier {
+        Identifier::from([0x11; 32])
+    }
+
+    fn dpns_contract() -> Arc<DataContract> {
+        let version = PlatformVersion::latest();
+        let mut contract =
+            dpp::tests::fixtures::get_dpns_data_contract_fixture(None, 0, version.protocol_version)
+                .data_contract_owned();
+        #[cfg(feature = "dpns-contract")]
+        contract.set_id(dpp::system_data_contracts::SystemDataContract::DPNS.id());
+        #[cfg(not(feature = "dpns-contract"))]
+        contract.set_id(
+            Identifier::from_string(
+                "GWRSAVFMjXx8HpQFaNJMqBV7MBgMK4br5UESsB4S31Ec",
+                dpp::platform_value::string_encoding::Encoding::Base58,
+            )
+            .expect("DPNS contract id"),
+        );
+        Arc::new(contract)
+    }
+
+    /// A parseable `domain` document for `label`, id byte `id`.
+    fn domain(id: u8, label: &str) -> (Identifier, Option<Document>) {
+        let records = vec![(
+            Value::Text("identity".to_string()),
+            Value::Identifier(identity().to_buffer()),
+        )];
+        let mut properties = BTreeMap::new();
+        properties.insert("label".to_string(), Value::Text(label.to_string()));
+        properties.insert(
+            "normalizedLabel".to_string(),
+            Value::Text(convert_to_homograph_safe_chars(label)),
+        );
+        properties.insert(
+            "normalizedParentDomainName".to_string(),
+            Value::Text("dash".to_string()),
+        );
+        properties.insert("records".to_string(), Value::Map(records));
+        let document_id = Identifier::from([id; 32]);
+        let document = Document::V0(DocumentV0 {
+            id: document_id,
+            owner_id: identity(),
+            properties,
+            ..Default::default()
+        });
+        (document_id, Some(document))
+    }
+
+    /// A document the username conversion skips (no `label`).
+    fn unparseable(id: u8) -> (Identifier, Option<Document>) {
+        let document_id = Identifier::from([id; 32]);
+        let document = Document::V0(DocumentV0 {
+            id: document_id,
+            owner_id: identity(),
+            ..Default::default()
+        });
+        (document_id, Some(document))
+    }
+
+    fn page(entries: Vec<(Identifier, Option<Document>)>) -> Documents {
+        entries.into_iter().collect()
+    }
+
+    /// Serves the DPNS contract (nothing else) so the loop's contract
+    /// lookup never reaches the network or a dump directory.
+    struct DpnsOnlyContextProvider(Arc<DataContract>);
+
+    impl dash_context_provider::ContextProvider for DpnsOnlyContextProvider {
+        fn get_data_contract(
+            &self,
+            id: &Identifier,
+            _platform_version: &PlatformVersion,
+        ) -> Result<Option<Arc<DataContract>>, crate::error::ContextProviderError> {
+            Ok((*id == self.0.id()).then(|| Arc::clone(&self.0)))
+        }
+
+        fn get_token_configuration(
+            &self,
+            _token_id: &Identifier,
+        ) -> Result<
+            Option<dpp::data_contract::associated_token::token_configuration::TokenConfiguration>,
+            crate::error::ContextProviderError,
+        > {
+            Ok(None)
+        }
+
+        fn get_quorum_public_key(
+            &self,
+            _quorum_type: u32,
+            _quorum_hash: [u8; 32],
+            _core_chain_locked_height: u32,
+        ) -> Result<[u8; 48], crate::error::ContextProviderError> {
+            Err(crate::error::ContextProviderError::Config(
+                "no quorum keys in this test".to_string(),
+            ))
+        }
+
+        fn get_platform_activation_height(
+            &self,
+        ) -> Result<dpp::prelude::CoreBlockHeight, crate::error::ContextProviderError> {
+            Ok(1)
+        }
+    }
+
+    /// Mock SDK serving the DPNS contract, with one expectation per
+    /// `(start_after, page)`; requests without an expectation fail.
+    async fn sdk_with_pages(pages: Vec<(Option<Identifier>, Documents)>) -> crate::Sdk {
+        sdk_with_pages_at(PlatformVersion::latest(), pages).await
+    }
+
+    /// [`sdk_with_pages`] pinned to `version`.
+    async fn sdk_with_pages_at(
+        version: &'static PlatformVersion,
+        pages: Vec<(Option<Identifier>, Documents)>,
+    ) -> crate::Sdk {
+        let contract = dpns_contract();
+        let mut sdk = SdkBuilder::new_mock()
+            .with_version(version)
+            .with_context_provider(DpnsOnlyContextProvider(Arc::clone(&contract)))
+            .build()
+            .expect("mock sdk");
+        for (start_after, documents) in pages {
+            let query =
+                identity_domains_page_query(Arc::clone(&contract), identity(), PAGE, start_after);
+            sdk.mock()
+                .expect_fetch_many::<Identifier, Document, _, Documents>(query, Some(documents))
+                .await
+                .expect("page expectation");
+        }
+        sdk
+    }
+
+    fn labels(usernames: &[DpnsUsername]) -> Vec<&str> {
+        usernames.iter().map(|u| u.label.as_str()).collect()
+    }
+
+    #[tokio::test]
+    async fn a_full_page_then_a_short_page_accumulates_and_completes() {
+        let sdk = sdk_with_pages(vec![
+            (None, page(vec![domain(1, "alice"), domain(2, "bob")])),
+            (
+                Some(Identifier::from([2; 32])),
+                page(vec![domain(3, "carol")]),
+            ),
+        ])
+        .await;
+
+        let (usernames, complete) = sdk
+            .get_all_dpns_usernames_by_identity(identity(), PAGE, 10)
+            .await
+            .expect("both pages answer");
+
+        assert_eq!(labels(&usernames), ["alice", "bob", "carol"]);
+        assert!(complete);
+    }
+
+    #[tokio::test]
+    async fn a_full_page_then_an_empty_page_completes() {
+        let sdk = sdk_with_pages(vec![
+            (None, page(vec![domain(1, "alice"), domain(2, "bob")])),
+            (Some(Identifier::from([2; 32])), page(Vec::new())),
+        ])
+        .await;
+
+        let (usernames, complete) = sdk
+            .get_all_dpns_usernames_by_identity(identity(), PAGE, 10)
+            .await
+            .expect("both pages answer");
+
+        assert_eq!(labels(&usernames), ["alice", "bob"]);
+        assert!(complete);
+    }
+
+    #[tokio::test]
+    async fn exhausting_the_page_budget_is_not_complete() {
+        let sdk = sdk_with_pages(vec![(
+            None,
+            page(vec![domain(1, "alice"), domain(2, "bob")]),
+        )])
+        .await;
+
+        let (usernames, complete) = sdk
+            .get_all_dpns_usernames_by_identity(identity(), PAGE, 1)
+            .await
+            .expect("the one page answers");
+
+        assert_eq!(labels(&usernames), ["alice", "bob"]);
+        assert!(
+            !complete,
+            "a full last page within the budget proves nothing"
+        );
+    }
+
+    #[tokio::test]
+    async fn an_error_on_a_later_page_propagates() {
+        // Only the first page has an expectation; the continuation fails.
+        let sdk = sdk_with_pages(vec![(
+            None,
+            page(vec![domain(1, "alice"), domain(2, "bob")]),
+        )])
+        .await;
+
+        let result = sdk
+            .get_all_dpns_usernames_by_identity(identity(), PAGE, 10)
+            .await;
+
+        let error = result.expect_err("a failed page must not read as complete");
+        assert!(
+            !matches!(error, crate::Error::ContextProviderError(_)),
+            "must fail on the unanswered second page, not on setup: {error:?}"
+        );
+    }
+
+    #[tokio::test]
+    async fn a_skipped_document_still_fills_the_page_and_moves_the_cursor() {
+        // Page 1 is full on raw entries even though one is unparseable; the
+        // continuation must start after that unparseable last document.
+        let sdk = sdk_with_pages(vec![
+            (None, page(vec![domain(1, "alice"), unparseable(2)])),
+            (
+                Some(Identifier::from([2; 32])),
+                page(vec![domain(3, "carol")]),
+            ),
+        ])
+        .await;
+
+        let (usernames, complete) = sdk
+            .get_all_dpns_usernames_by_identity(identity(), PAGE, 10)
+            .await
+            .expect("both pages answer");
+
+        assert_eq!(labels(&usernames), ["alice", "carol"]);
+        assert!(complete);
+    }
+
+    /// Protocol version 13 cannot continue inside the equality bucket, so a
+    /// full first page is all the loop reads, and it is not complete. Only
+    /// the first page has an expectation: a continuation would fail.
+    #[tokio::test]
+    async fn a_full_page_on_the_v0_lowering_stops_and_is_not_complete() {
+        let v13 = PlatformVersion::get(13).expect("protocol version 13");
+        let sdk = sdk_with_pages_at(
+            v13,
+            vec![(None, page(vec![domain(1, "alice"), domain(2, "bob")]))],
+        )
+        .await;
+
+        let (usernames, complete) = sdk
+            .get_all_dpns_usernames_by_identity(identity(), PAGE, 10)
+            .await
+            .expect("only the first page is requested");
+
+        assert_eq!(labels(&usernames), ["alice", "bob"]);
+        assert!(!complete, "a v0 continuation would drop the rest");
+    }
+
+    /// A short first page needs no cursor, so it is complete on any version.
+    #[tokio::test]
+    async fn a_short_first_page_on_the_v0_lowering_is_complete() {
+        let v13 = PlatformVersion::get(13).expect("protocol version 13");
+        let sdk = sdk_with_pages_at(v13, vec![(None, page(vec![domain(1, "alice")]))]).await;
+
+        let (usernames, complete) = sdk
+            .get_all_dpns_usernames_by_identity(identity(), PAGE, 10)
+            .await
+            .expect("the one page answers");
+
+        assert_eq!(labels(&usernames), ["alice"]);
+        assert!(complete);
+    }
+}
diff --git a/packages/swift-sdk/Sources/SwiftDashSDK/Persistence/Models/PersistentIdentity.swift b/packages/swift-sdk/Sources/SwiftDashSDK/Persistence/Models/PersistentIdentity.swift
index 72e637a815c..df7543d756c 100644
--- a/packages/swift-sdk/Sources/SwiftDashSDK/Persistence/Models/PersistentIdentity.swift
+++ b/packages/swift-sdk/Sources/SwiftDashSDK/Persistence/Models/PersistentIdentity.swift
@@ -246,15 +246,14 @@ public final class PersistentIdentity {
         publicKeys.compactMap { $0.toIdentityPublicKey() }
     }
 
-    /// User-facing short name. Priority: `alias` → `mainDpnsName`
-    /// → `dpnsName` → truncated hex id. Mirrors the old
-    /// `IdentityModel.displayName` extension so views that read
-    /// this don't change behavior post-migration.
+    /// User-facing short name. Priority: `alias` → `ownedMainDpnsName`
+    /// (the user's pick while it is still owned) → `dpnsName` → truncated
+    /// hex id.
     public var displayName: String {
         if let alias = alias, !alias.isEmpty {
             return alias
         }
-        if let mainDpnsName = mainDpnsName, !mainDpnsName.isEmpty {
+        if let mainDpnsName = ownedMainDpnsName {
             return mainDpnsName
         }
         if let dpnsName = dpnsName, !dpnsName.isEmpty {
@@ -263,6 +262,48 @@ public final class PersistentIdentity {
         return String(identityIdString.prefix(12)) + "..."
     }
 
+    /// The user's `mainDpnsName` pick while the identity still owns it — the
+    /// value every display surface should use. `mainDpnsName` itself is the
+    /// stored selection, kept as written: a sold or transferred name stays
+    /// there, and is skipped here once its label row is no longer owned (the
+    /// persister keeps the picked name's row for exactly this check). With
+    /// no label rows at all, the identity has not been hydrated yet and the
+    /// pick is trusted — unless the label's row, unique per network, now
+    /// belongs to another identity: a transfer within the wallet rebinds it,
+    /// leaving the old owner with no rows. If that lookup fails, the pick is
+    /// not shown.
+    public var ownedMainDpnsName: String? {
+        guard let mainDpnsName, !mainDpnsName.isEmpty else { return nil }
+        let normalized = PersistentDPNSName.normalize(mainDpnsName)
+        let names = dpnsNames
+        guard names.isEmpty else {
+            return names.contains { $0.isOwned && $0.normalizedLabel == normalized } ? mainDpnsName : nil
+        }
+        guard let context = modelContext else { return mainDpnsName }
+        let networkRaw = networkRaw
+        let parent = PersistentDPNSName.normalize("dash")
+        var descriptor = FetchDescriptor<PersistentDPNSName>(
+            predicate: #Predicate {
+                $0.networkRaw == networkRaw
+                    && $0.normalizedParentDomainName == parent
+                    && $0.normalizedLabel == normalized
+            }
+        )
+        descriptor.fetchLimit = 1
+        // A failed lookup proves nothing about ownership, so it is not
+        // read as "no other owner".
+        let row: PersistentDPNSName?
+        do {
+            row = try context.fetch(descriptor).first
+        } catch {
+            return nil
+        }
+        if let row, row.identity.identityId != identityId {
+            return nil
+        }
+        return mainDpnsName
+    }
+
     public var identityTypeEnum: IdentityType {
         IdentityType(rawValue: identityType) ?? .user
     }
diff --git a/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletPersistenceHandler.swift b/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletPersistenceHandler.swift
index d9ffdea6697..474e9026c0e 100644
--- a/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletPersistenceHandler.swift
+++ b/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletPersistenceHandler.swift
@@ -3711,14 +3711,21 @@ public final class PlatformWalletPersistenceHandler: @unchecked Sendable {
         let previouslyAssociatedRows =
             (try? backgroundContext.fetch(ownedRowsDescriptor)) ?? Array(identityRow.dpnsNames)
 
+        let pickedLabel = identityRow.mainDpnsName.map(PersistentDPNSName.normalize)
         for row in previouslyAssociatedRows
         where !canonicalLabels.contains(row.normalizedLabel) {
-            row.isOwned = false
-            row.lastUpdated = Date()
-            if row.documentIdBase58 == nil {
+            if row.documentIdBase58 == nil && row.normalizedLabel != pickedLabel {
                 // No marketplace history is attached, so this is only a stale
-                // label-cache row and can be removed entirely.
+                // label-cache row and can be removed entirely. The picked
+                // name's row stays, marked not owned, so readers can tell a
+                // departed pick from an identity that has no rows yet.
                 backgroundContext.delete(row)
+            } else if row.isOwned {
+                // Only the departure itself is a change: a row already
+                // retained as not owned is left untouched by every later
+                // snapshot that omits it, so it is not re-dirtied each time.
+                row.isOwned = false
+                row.lastUpdated = Date()
             }
         }
 
@@ -3755,6 +3762,25 @@ public final class PlatformWalletPersistenceHandler: @unchecked Sendable {
                 // transfers, and the unique constraint is per-network,
                 // so the row stays but the owner pointer moves.
                 if existing.identity !== identityRow {
+                    // The rebind is positive evidence the name left the
+                    // previous owner, so its selections of that name go too
+                    // — unlike a label merely missing from a snapshot. Once
+                    // the row moves, the old owner has nothing left to check
+                    // a stale pick against.
+                    let previousOwner = existing.identity
+                    if previousOwner.mainDpnsName.map(PersistentDPNSName.normalize) == normalizedLabel {
+                        previousOwner.mainDpnsName = nil
+                    }
+                    if previousOwner.dpnsName.map(PersistentDPNSName.normalize) == normalizedLabel {
+                        // Fall back to another name the old owner still
+                        // owns, as the snapshot's own fallback does for the
+                        // identity it describes — no snapshot for the old
+                        // owner may follow to repair it.
+                        previousOwner.dpnsName = previousOwner.dpnsNames
+                            .filter { $0.isOwned && $0.normalizedLabel != normalizedLabel }
+                            .sorted { ($0.acquiredAt, $0.label) < ($1.acquiredAt, $1.label) }
+                            .first?.label
+                    }
                     existing.identity = identityRow
                     existing.lastUpdated = Date()
                 }
@@ -3769,13 +3795,40 @@ public final class PlatformWalletPersistenceHandler: @unchecked Sendable {
             }
         }
 
-        let fallbackLabel = names.first?.label
-        if let selected = identityRow.mainDpnsName,
-           !canonicalLabels.contains(PersistentDPNSName.normalize(selected)) {
-            identityRow.mainDpnsName = fallbackLabel
+        // `mainDpnsName` is the user's pick and is never rewritten here: a
+        // snapshot can be momentarily incomplete (a cold start adds names
+        // before the in-memory list is whole), and resetting the pick on
+        // one lost it for good. Readers skip a pick that is no longer owned
+        // — see `PersistentIdentity.displayName`.
+        //
+        // A pick the snapshot omits and that has no row anywhere on the
+        // network (stored before rows existed) gets a not-owned row, so the
+        // omission is recorded instead of the pick being trusted as
+        // unhydrated. A later snapshot carrying the name owns the row again.
+        if let pick = identityRow.mainDpnsName, !pick.isEmpty,
+           let pickedLabel, !canonicalLabels.contains(pickedLabel) {
+            let pickDescriptor = FetchDescriptor<PersistentDPNSName>(
+                predicate: #Predicate {
+                    $0.networkRaw == networkRaw
+                        && $0.normalizedParentDomainName == normalizedParentDomainName
+                        && $0.normalizedLabel == pickedLabel
+                }
+            )
+            if ((try? backgroundContext.fetch(pickDescriptor)) ?? []).isEmpty {
+                let row = PersistentDPNSName(
+                    identity: identityRow,
+                    label: pick,
+                    parentDomainName: parentDomainName
+                )
+                row.isOwned = false
+                backgroundContext.insert(row)
+            }
         }
-        if let displayed = identityRow.dpnsName,
-           !canonicalLabels.contains(PersistentDPNSName.normalize(displayed)) {
+
+        // The SDK's own display cache: kept on an owned name, including when
+        // it was never set (a pick alone does not fill it).
+        let fallbackLabel = names.first?.label
+        if identityRow.dpnsName.map({ !canonicalLabels.contains(PersistentDPNSName.normalize($0)) }) ?? true {
             identityRow.dpnsName = fallbackLabel
         }
     }
@@ -8160,12 +8213,33 @@ public final class PlatformWalletPersistenceHandler: @unchecked Sendable {
             // identity changeset path.
             entry.status = 0
 
-            // DPNS names — currently empty. Wiring is here so a
-            // future query against `PersistentDpnsName` rows (or a
-            // dedicated array column on the identity) drops in
-            // without touching the FFI plumbing.
-            entry.dpns_names = nil
-            entry.dpns_names_count = 0
+            // DPNS names — the identity's OWNED label rows, so Rust's
+            // `ManagedIdentity.dpns_names` starts with every name already
+            // known instead of empty. A capped (partial) username fetch
+            // then only adds to that list, and no snapshot emitted before
+            // the next complete fetch carries a truncated owned set that
+            // `upsertDPNSNames` would read as departures. Rows retained as
+            // not owned (departed names, a picked name the last snapshot
+            // omitted) are not restored. Order: oldest acquisition first,
+            // then label, so the restored list is deterministic.
+            let ownedLabels = identity.dpnsNames
+                .filter { $0.isOwned }
+                .sorted { ($0.acquiredAt, $0.label) < ($1.acquiredAt, $1.label) }
+                .map(\.label)
+            if ownedLabels.isEmpty {
+                entry.dpns_names = nil
+                entry.dpns_names_count = 0
+            } else {
+                let labelArray = UnsafeMutablePointer<UnsafePointer<CChar>?>.allocate(
+                    capacity: ownedLabels.count
+                )
+                for (i, label) in ownedLabels.enumerated() {
+                    labelArray[i] = UnsafePointer(duplicateCString(label, allocation: allocation))
+                }
+                allocation.cStringPointerArrays.append((labelArray, ownedLabels.count))
+                entry.dpns_names = UnsafePointer(labelArray)
+                entry.dpns_names_count = UInt(ownedLabels.count)
+            }
             entry.contested_dpns_names = nil
             entry.contested_dpns_names_count = 0
 
diff --git a/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Views/DashPay/CreateInvitationSheet.swift b/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Views/DashPay/CreateInvitationSheet.swift
index 0b37a88434c..dc4ce1103fd 100644
--- a/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Views/DashPay/CreateInvitationSheet.swift
+++ b/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Views/DashPay/CreateInvitationSheet.swift
@@ -53,7 +53,7 @@ struct CreateInvitationSheet: View {
     /// The inviter's DPNS username, if registered. The contact-bootstrap can only
     /// be offered when the inviter has a username to advertise in the link.
     private var username: String? {
-        let name = (identity.mainDpnsName ?? identity.dpnsName)?
+        let name = (identity.ownedMainDpnsName ?? identity.dpnsName)?
             .trimmingCharacters(in: .whitespacesAndNewlines)
         return (name?.isEmpty == false) ? name : nil
     }
diff --git a/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Views/DashPay/DashPayProfileView.swift b/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Views/DashPay/DashPayProfileView.swift
index b3b9aad1a98..9b3dc6e3e8d 100644
--- a/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Views/DashPay/DashPayProfileView.swift
+++ b/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Views/DashPay/DashPayProfileView.swift
@@ -26,7 +26,7 @@ struct DashPayProfileView: View {
            !name.isEmpty {
             return name
         }
-        if let dpns = identity.mainDpnsName ?? identity.dpnsName {
+        if let dpns = identity.ownedMainDpnsName ?? identity.dpnsName {
             return dpns
         }
         return String(identity.identityIdBase58.prefix(12)) + "…"
@@ -45,7 +45,7 @@ struct DashPayProfileView: View {
                         Text(displayName)
                             .font(.title2)
                             .fontWeight(.semibold)
-                        if let dpns = identity.mainDpnsName ?? identity.dpnsName {
+                        if let dpns = identity.ownedMainDpnsName ?? identity.dpnsName {
                             Text(dpns)
                                 .font(.subheadline)
                                 .foregroundColor(.blue)
@@ -158,7 +158,7 @@ struct DashPayProfileView: View {
         // on-chain when the cache is empty (imported/restored identities carry
         // the name on-chain but not in the local field). If no name is
         // registered at all, the Rust call surfaces a clear error.
-        let username = (identity.mainDpnsName ?? identity.dpnsName)?
+        let username = (identity.ownedMainDpnsName ?? identity.dpnsName)?
             .trimmingCharacters(in: .whitespacesAndNewlines) ?? ""
         do {
             let uri = try await wallet.buildAutoAcceptQR(
diff --git a/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Views/DashPay/DashPayTabView.swift b/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Views/DashPay/DashPayTabView.swift
index 1eb043a4c22..e7855efb305 100644
--- a/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Views/DashPay/DashPayTabView.swift
+++ b/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Views/DashPay/DashPayTabView.swift
@@ -339,7 +339,7 @@ struct DashPayTabView: View {
                         // resolved set so this prompt hides on the next render.
                         // The Rust register path only upserts `PersistentDPNSName`
                         // relationship rows — not the scalar `dpnsName` /
-                        // `mainDpnsName` this prompt (and the header) read — so
+                        // `ownedMainDpnsName` this prompt (and the header) read — so
                         // without this the CTA would linger until the next
                         // identity switch or app-foreground re-check.
                         RegisterNameView(identity: identity, onRegistered: { label in
@@ -609,7 +609,7 @@ struct DashPayTabView: View {
 
     /// Menu rows show "DPNS name → truncated id".
     private func pickerLabel(for identity: PersistentIdentity) -> String {
-        if let name = identity.mainDpnsName ?? identity.dpnsName, !name.isEmpty {
+        if let name = identity.ownedMainDpnsName ?? identity.dpnsName, !name.isEmpty {
             return name
         }
         if let alias = identity.alias, !alias.isEmpty {
@@ -636,7 +636,7 @@ struct DashPayTabView: View {
                         Text(headerDisplayName(identity: identity, profile: profile))
                             .font(.headline)
                             .foregroundColor(.primary)
-                        if let dpns = identity.mainDpnsName ?? identity.dpnsName {
+                        if let dpns = identity.ownedMainDpnsName ?? identity.dpnsName {
                             Text(dpns)
                                 .font(.caption)
                                 .foregroundColor(.secondary)
@@ -705,7 +705,7 @@ struct DashPayTabView: View {
     /// display name is cosmetic and not searchable.
     @ViewBuilder
     private func usernamePromptCard(identity: PersistentIdentity) -> some View {
-        let hasName = (identity.mainDpnsName ?? identity.dpnsName)
+        let hasName = (identity.ownedMainDpnsName ?? identity.dpnsName)
             .map { !$0.isEmpty } ?? false
         if !hasName, usernameResolvedIds.contains(identity.identityId) {
             Button {
@@ -783,7 +783,7 @@ struct DashPayTabView: View {
            !name.isEmpty {
             return name
         }
-        if let dpns = identity.mainDpnsName ?? identity.dpnsName {
+        if let dpns = identity.ownedMainDpnsName ?? identity.dpnsName {
             return dpns
         }
         return String(identity.identityIdBase58.prefix(12)) + "…"
@@ -833,7 +833,7 @@ struct DashPayTabView: View {
     /// DPNS fetch in `IdentitiesView`.
     private func resolveUsernameIfNeeded(for identity: PersistentIdentity?) async {
         guard let identity else { return }
-        if let name = identity.mainDpnsName ?? identity.dpnsName, !name.isEmpty {
+        if let name = identity.ownedMainDpnsName ?? identity.dpnsName, !name.isEmpty {
             return
         }
         guard let sdk = appState.sdk else { return }
diff --git a/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Views/IdentitiesView.swift b/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Views/IdentitiesView.swift
index 461e2f70446..1461c49bbfc 100644
--- a/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Views/IdentitiesView.swift
+++ b/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Views/IdentitiesView.swift
@@ -50,10 +50,10 @@ struct IdentityRow: View {
                         HStack(spacing: 4) {
                             Text(identity.displayName)
                                 .font(.headline)
-                                .foregroundColor(identity.mainDpnsName != nil || identity.dpnsName != nil ? .blue : .primary)
+                                .foregroundColor(identity.ownedMainDpnsName != nil || identity.dpnsName != nil ? .blue : .primary)
 
                             // Show star icon if this is the selected main name
-                            if identity.mainDpnsName != nil {
+                            if identity.ownedMainDpnsName != nil {
                                 Image(systemName: "star.fill")
                                     .font(.caption)
                                     .foregroundColor(.yellow)
@@ -61,7 +61,7 @@ struct IdentityRow: View {
                         }
 
                         // Show alias as subtitle if we're displaying a DPNS name
-                        if (identity.mainDpnsName != nil || identity.dpnsName != nil),
+                        if (identity.ownedMainDpnsName != nil || identity.dpnsName != nil),
                            let alias = identity.alias {
                             Text(alias)
                                 .font(.caption)
@@ -167,7 +167,7 @@ struct IdentityRow: View {
             // Fetch a DPNS name if we don't already have one — but
             // only once (silent failure), since not every identity
             // has a DPNS name and the request can 404.
-            if identity.dpnsName == nil && identity.mainDpnsName == nil {
+            if identity.dpnsName == nil && identity.ownedMainDpnsName == nil {
                 if let usernames = try? await sdk.dpnsGetUsername(
                     identityId: identity.identityIdBase58,
                     limit: 1
diff --git a/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Views/IdentityDetailView.swift b/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Views/IdentityDetailView.swift
index 95f8479e58c..682c3393103 100644
--- a/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Views/IdentityDetailView.swift
+++ b/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Views/IdentityDetailView.swift
@@ -111,7 +111,7 @@ struct IdentityDetailView: View {
                         }
 
                     // Show the main name if selected, otherwise show first registered name
-                    if let mainName = identity.mainDpnsName {
+                    if let mainName = identity.ownedMainDpnsName {
                         HStack {
                             Label(mainName, systemImage: "star.fill")
                                 .font(.subheadline)
diff --git a/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Views/StorageModelListViews.swift b/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Views/StorageModelListViews.swift
index b4627318cf0..23bc60dd8c3 100644
--- a/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Views/StorageModelListViews.swift
+++ b/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Views/StorageModelListViews.swift
@@ -321,7 +321,7 @@ struct PublicKeyStorageListView: View {
     /// name, then alias, then a short identity id — avoids the
     /// generic "Identity" placeholder except for truly empty rows.
     private func identityDisplayName(_ identity: PersistentIdentity) -> String {
-        if let name = identity.mainDpnsName, !name.isEmpty {
+        if let name = identity.ownedMainDpnsName, !name.isEmpty {
             return name
         }
         if let name = identity.dpnsName, !name.isEmpty {
diff --git a/packages/swift-sdk/SwiftTests/SwiftDashSDKTests/DpnsMarketplacePersistenceTests.swift b/packages/swift-sdk/SwiftTests/SwiftDashSDKTests/DpnsMarketplacePersistenceTests.swift
index 0260b77aa4f..eca3e59b8e6 100644
--- a/packages/swift-sdk/SwiftTests/SwiftDashSDKTests/DpnsMarketplacePersistenceTests.swift
+++ b/packages/swift-sdk/SwiftTests/SwiftDashSDKTests/DpnsMarketplacePersistenceTests.swift
@@ -1,5 +1,6 @@
 import XCTest
 import SwiftData
+import DashSDKFFI
 @testable import SwiftDashSDK
 
 final class DpnsMarketplacePersistenceTests: XCTestCase {
@@ -56,6 +57,42 @@ final class DpnsMarketplacePersistenceTests: XCTestCase {
         XCTAssertTrue(handler.endChangeset(walletId: walletId, success: true))
     }
 
+    /// A cold restore hands Rust every OWNED name the store knows, so the
+    /// in-memory list is never empty/truncated before the first complete
+    /// fetch: a capped fetch can then only add, and no snapshot reads an
+    /// owned pick outside the fetched prefix as departed. Rows retained as
+    /// not owned stay out of the restored list.
+    func testColdRestoreCarriesEveryOwnedNameToRust() throws {
+        let context = ModelContext(container)
+        let wallet = PersistentWallet(walletId: walletId, network: .testnet)
+        context.insert(wallet)
+        let account = PersistentAccount(
+            wallet: wallet, accountType: 0, accountIndex: 0, accountTypeName: "Standard")
+        account.accountExtendedPubKeyBytes = Data(repeating: 0x30, count: 78)
+        context.insert(account)
+        let identity = PersistentIdentity(
+            identityId: ownerId, mainDpnsName: "Carol", network: .testnet)
+        identity.wallet = wallet
+        context.insert(identity)
+        for (label, acquiredAt, owned) in [
+            ("Bob", UInt64(20), true), ("Alice", 10, true), ("Carol", 30, true), ("Dave", 5, false),
+        ] {
+            let row = PersistentDPNSName(identity: identity, label: label, acquiredAt: acquiredAt)
+            row.isOwned = owned
+            context.insert(row)
+        }
+        try context.save()
+
+        let loaded = handler.loadWalletList()
+        XCTAssertFalse(loaded.errored)
+        let entries = try XCTUnwrap(loaded.entries)
+        defer { handler.loadWalletListFree(entries: UnsafeRawPointer(entries)) }
+        let restored = try XCTUnwrap(entries[0].identities)[0]
+        let names = try XCTUnwrap(restored.dpns_names)
+        let labels = (0..<Int(restored.dpns_names_count)).map { String(cString: names[$0]!) }
+        XCTAssertEqual(labels, ["Alice", "Bob", "Carol"])
+    }
+
     func testMarketplaceColumnsHaveMigrationSafeDefaults() {
         let identity = PersistentIdentity(
             identityId: ownerId,
@@ -93,7 +130,8 @@ final class DpnsMarketplacePersistenceTests: XCTestCase {
         try context.save()
 
         // Alice leaves the canonical owned set. Keep its marketplace history,
-        // while Bob becomes the fallback display/main name.
+        // while Bob becomes the fallback display name. The main-name pick
+        // stays as the user wrote it; readers skip it once it is not owned.
         handler.beginChangeset(walletId: walletId)
         handler.persistIdentities(
             walletId: walletId,
@@ -130,11 +168,13 @@ final class DpnsMarketplacePersistenceTests: XCTestCase {
         )
         XCTAssertEqual(owned.map(\.label), ["Bob"])
         var identity = try XCTUnwrap(PersistentIdentity.fetch(in: readContext, identityId: ownerId))
-        XCTAssertEqual(identity.mainDpnsName, "Bob")
+        XCTAssertEqual(identity.mainDpnsName, "Alice")
+        XCTAssertNil(identity.ownedMainDpnsName)
         XCTAssertEqual(identity.dpnsName, "Bob")
+        XCTAssertEqual(identity.displayName, "Bob")
 
         // An empty canonical snapshot removes Bob (cache-only), keeps Alice's
-        // sold history, and clears stale scalar selections.
+        // sold history, and clears the stale display scalar.
         applyIdentitySnapshot(id: ownerId, names: [])
         readContext = ModelContext(container)
         owned = try readContext.fetch(
@@ -152,10 +192,124 @@ final class DpnsMarketplacePersistenceTests: XCTestCase {
         XCTAssertEqual(allRows.first?.documentUpdatedAtMs, 12)
         XCTAssertEqual(allRows.first?.documentTransferredAtMs, 13)
         identity = try XCTUnwrap(PersistentIdentity.fetch(in: readContext, identityId: ownerId))
-        XCTAssertNil(identity.mainDpnsName)
+        XCTAssertNil(identity.ownedMainDpnsName)
         XCTAssertNil(identity.dpnsName)
     }
 
+    /// A pick with no marketplace history that leaves the owned set keeps its
+    /// row (not owned), so an identity left with no other names does not read
+    /// as unhydrated and resurface the departed pick.
+    func testDepartedCacheOnlyPickIsNotDisplayed() throws {
+        applyIdentitySnapshot(id: ownerId, names: [("Alice", 10)])
+        let context = ModelContext(container)
+        XCTAssertTrue(PersistentIdentity.updateMainDpnsName(
+            in: context, identityId: ownerId, mainDpnsName: "Alice"))
+        try context.save()
+
+        applyIdentitySnapshot(id: ownerId, names: [])
+
+        let readContext = ModelContext(container)
+        let identity = try XCTUnwrap(PersistentIdentity.fetch(in: readContext, identityId: ownerId))
+        XCTAssertEqual(identity.mainDpnsName, "Alice")
+        XCTAssertEqual(identity.dpnsNames.map(\.label), ["Alice"])
+        XCTAssertEqual(identity.dpnsNames.first?.isOwned, false)
+        XCTAssertNil(identity.ownedMainDpnsName)
+        XCTAssertNotEqual(identity.displayName, "Alice")
+
+        // Owning it again brings the pick back.
+        applyIdentitySnapshot(id: ownerId, names: [("Alice", 10)])
+        let reread = try XCTUnwrap(PersistentIdentity.fetch(in: ModelContext(container), identityId: ownerId))
+        XCTAssertEqual(reread.ownedMainDpnsName, "Alice")
+    }
+
+    /// A retained not-owned row is written once, when it departs: later
+    /// snapshots that still omit it leave it untouched instead of re-dirtying
+    /// it with a fresh `lastUpdated`.
+    func testRetainedNotOwnedPickRowIsNotRewrittenByLaterSnapshots() throws {
+        applyIdentitySnapshot(id: ownerId, names: [("Alice", 10), ("Bob", 20)])
+        let context = ModelContext(container)
+        XCTAssertTrue(PersistentIdentity.updateMainDpnsName(
+            in: context, identityId: ownerId, mainDpnsName: "Alice"))
+        try context.save()
+
+        applyIdentitySnapshot(id: ownerId, names: [("Bob", 20)])
+        let departedAt = try XCTUnwrap(
+            PersistentIdentity.fetch(in: ModelContext(container), identityId: ownerId)?
+                .dpnsNames.first { $0.label == "Alice" }?.lastUpdated)
+
+        applyIdentitySnapshot(id: ownerId, names: [("Bob", 20)])
+        applyIdentitySnapshot(id: ownerId, names: [("Bob", 20)])
+
+        let identity = try XCTUnwrap(PersistentIdentity.fetch(in: ModelContext(container), identityId: ownerId))
+        let alice = try XCTUnwrap(identity.dpnsNames.first { $0.label == "Alice" })
+        XCTAssertFalse(alice.isOwned)
+        XCTAssertEqual(alice.lastUpdated, departedAt)
+        XCTAssertNil(identity.ownedMainDpnsName)
+    }
+
+    /// A departed pick with no display cache (`dpnsName` never set — a pick
+    /// alone does not fill it) still falls back to another owned name.
+    func testDisplayFallsBackToAnOwnedNameWhenTheDisplayCacheIsNil() throws {
+        applyIdentitySnapshot(id: ownerId, names: [("Alice", 10), ("Bob", 20)])
+        let context = ModelContext(container)
+        XCTAssertTrue(PersistentIdentity.updateMainDpnsName(
+            in: context, identityId: ownerId, mainDpnsName: "Alice"))
+        try XCTUnwrap(PersistentIdentity.fetch(in: context, identityId: ownerId)).dpnsName = nil
+        try context.save()
+
+        applyIdentitySnapshot(id: ownerId, names: [("Bob", 20)])
+
+        let identity = try XCTUnwrap(PersistentIdentity.fetch(in: ModelContext(container), identityId: ownerId))
+        XCTAssertEqual(identity.mainDpnsName, "Alice")
+        XCTAssertNil(identity.ownedMainDpnsName)
+        XCTAssertEqual(identity.dpnsName, "Bob")
+        XCTAssertEqual(identity.displayName, "Bob")
+    }
+
+    /// A pick stored without any name row (older data) is not trusted as
+    /// unhydrated once an authoritative snapshot omits it, and comes back
+    /// when a snapshot carries it again.
+    func testScalarOnlyPickOmittedBySnapshotIsNotDisplayed() throws {
+        let context = ModelContext(container)
+        context.insert(PersistentIdentity(
+            identityId: ownerId,
+            isLocal: false,
+            mainDpnsName: "Alice",
+            network: .testnet
+        ))
+        try context.save()
+
+        applyIdentitySnapshot(id: ownerId, names: [])
+        var identity = try XCTUnwrap(PersistentIdentity.fetch(in: ModelContext(container), identityId: ownerId))
+        XCTAssertEqual(identity.mainDpnsName, "Alice")
+        XCTAssertNil(identity.ownedMainDpnsName)
+        XCTAssertNotEqual(identity.displayName, "Alice")
+
+        applyIdentitySnapshot(id: ownerId, names: [("Alice", 10)])
+        identity = try XCTUnwrap(PersistentIdentity.fetch(in: ModelContext(container), identityId: ownerId))
+        XCTAssertEqual(identity.ownedMainDpnsName, "Alice")
+        XCTAssertEqual(identity.displayName, "Alice")
+    }
+
+    /// A snapshot that momentarily lacks the picked name (a cold start adds
+    /// names before the in-memory list is whole) must not replace the pick.
+    func testMainNamePickSurvivesAnIncompleteSnapshot() throws {
+        applyIdentitySnapshot(id: ownerId, names: [("Alice", 10), ("Bob", 20)])
+        let context = ModelContext(container)
+        XCTAssertTrue(PersistentIdentity.updateMainDpnsName(
+            in: context, identityId: ownerId, mainDpnsName: "Bob"))
+        try context.save()
+
+        applyIdentitySnapshot(id: ownerId, names: [("Alice", 30)])
+        applyIdentitySnapshot(id: ownerId, names: [("Alice", 30), ("Bob", 30)])
+
+        let readContext = ModelContext(container)
+        let identity = try XCTUnwrap(PersistentIdentity.fetch(in: readContext, identityId: ownerId))
+        XCTAssertEqual(identity.mainDpnsName, "Bob")
+        XCTAssertEqual(identity.ownedMainDpnsName, "Bob")
+        XCTAssertEqual(identity.displayName, "Bob")
+    }
+
     func testMarketplaceCallbackCannotRestoreOwnershipRemovedByCanonicalSnapshot() throws {
         let documentId = Data(repeating: 0x35, count: 32).toBase58String()
         let context = ModelContext(container)
@@ -211,10 +365,44 @@ final class DpnsMarketplacePersistenceTests: XCTestCase {
         ).isEmpty)
     }
 
+    /// A transfer clears the previous owner's selection of the moved name;
+    /// its display cache falls back to another name it still owns, since no
+    /// snapshot for the previous owner may follow.
+    func testPreviousOwnerFallsBackToAnotherOwnedNameAfterTransfer() throws {
+        let context = ModelContext(container)
+        let oldOwner = PersistentIdentity(
+            identityId: ownerId,
+            isLocal: false,
+            dpnsName: "Alice",
+            mainDpnsName: "Alice",
+            network: .testnet
+        )
+        let nextOwner = PersistentIdentity(identityId: nextOwnerId, isLocal: false, network: .testnet)
+        context.insert(oldOwner)
+        context.insert(nextOwner)
+        context.insert(PersistentDPNSName(identity: oldOwner, label: "Alice", acquiredAt: 10))
+        context.insert(PersistentDPNSName(identity: oldOwner, label: "Bob", acquiredAt: 20))
+        try context.save()
+
+        applyIdentitySnapshot(id: nextOwnerId, names: [("Alice", 30)])
+
+        let readContext = ModelContext(container)
+        let previous = try XCTUnwrap(PersistentIdentity.fetch(in: readContext, identityId: ownerId))
+        XCTAssertNil(previous.mainDpnsName)
+        XCTAssertEqual(previous.dpnsName, "Bob")
+        XCTAssertEqual(previous.displayName, "Bob")
+        XCTAssertEqual(previous.dpnsNames.map(\.label), ["Bob"])
+    }
+
     func testSameWalletTransferRebindsSingleCanonicalRowToNewOwner() throws {
         let documentId = Data(repeating: 0x36, count: 32).toBase58String()
         let context = ModelContext(container)
-        let oldOwner = PersistentIdentity(identityId: ownerId, isLocal: false, network: .testnet)
+        let oldOwner = PersistentIdentity(
+            identityId: ownerId,
+            isLocal: false,
+            mainDpnsName: "Alice",
+            network: .testnet
+        )
         let nextOwner = PersistentIdentity(
             identityId: nextOwnerId,
             isLocal: false,
@@ -276,6 +464,25 @@ final class DpnsMarketplacePersistenceTests: XCTestCase {
                 predicate: PersistentDPNSName.predicate(identityId: nextOwnerId)
             )
         ).map(\.label), ["Alice"])
+
+        // The rebind clears the old owner's pick of the transferred name.
+        let previous = try XCTUnwrap(PersistentIdentity.fetch(in: readContext, identityId: ownerId))
+        XCTAssertNil(previous.mainDpnsName)
+        XCTAssertTrue(previous.dpnsNames.isEmpty)
+        XCTAssertNil(previous.ownedMainDpnsName)
+        XCTAssertNotEqual(previous.displayName, "Alice")
+
+        // Even once the recipient's row is gone (its wallet deleted), the old
+        // owner has no stale pick left to resurface.
+        let deleteContext = ModelContext(container)
+        for row in try deleteContext.fetch(FetchDescriptor<PersistentDPNSName>()) {
+            deleteContext.delete(row)
+        }
+        try deleteContext.save()
+        let afterDelete = try XCTUnwrap(
+            PersistentIdentity.fetch(in: ModelContext(container), identityId: ownerId))
+        XCTAssertNil(afterDelete.ownedMainDpnsName)
+        XCTAssertNotEqual(afterDelete.displayName, "Alice")
     }
 
     func testMarketplaceRemovalClearsOnlyMarketplaceColumns() throws {