Update/staking hotkeys limitation for coldkey swap

bittensor/core/async_subtensor.py

@@ -4468,6 +4468,36 @@ async def get_stake_weight(
         )
         return [u16_normalized_float(w) for w in cast(list[int], result or [])]
 
+    async def get_staking_hotkeys(
+        self,
+        coldkey_ss58: str,
+        block: Optional[int] = None,
+        block_hash: Optional[str] = None,
+        reuse_block: bool = False,
+    ) -> list[str]:
+        """
+        Retrieves the hotkeys that have staked for a given coldkey.
+
+        Parameters:
+            coldkey_ss58: The SS58 address of the coldkey.
+            block: The block number at which to query the stake information.
+            block_hash: The hash of the blockchain block number for the query.
+            reuse_block: Whether to reuse the last-used block hash.
+
+        Returns:
+            A list of hotkey SS58 addresses that have staked for the given coldkey.
+        """
+        block_hash = await self.determine_block_hash(
+            block=block, block_hash=block_hash, reuse_block=reuse_block
+        )
+        result = await self.substrate.query(
+            module="SubtensorModule",
+            storage_function="StakingHotkeys",
+            params=[coldkey_ss58],
+            block_hash=block_hash,
+        )
+        return [decode_account_id(hotkey[0]) for hotkey in result or []]
+
     async def get_start_call_delay(
         self,
         block: Optional[int] = None,

bittensor/core/extrinsics/asyncex/coldkey_swap.py

@@ -60,6 +60,7 @@ async def announce_coldkey_swap_extrinsic(
         - A swap cost is charged when making the first announcement (not when reannouncing).
         - After making an announcement, all transactions from the coldkey are blocked except for `swap_coldkey_announced`.
         - The swap can only be executed after the delay period has passed (check via `get_coldkey_swap_announcement`).
+        - The destination coldkey cannot have any staking hotkeys. It must be completely new without any staking activity.
         - See: <https://docs.learnbittensor.org/keys/coldkey-swap>
     """
     try:
@@ -68,6 +69,17 @@ async def announce_coldkey_swap_extrinsic(
         ).success:
             return unlocked
 
+        staking_hotkeys = await subtensor.get_staking_hotkeys(new_coldkey_ss58)
+        if staking_hotkeys:
+            error_msg = "Destination coldkey cannot have any staking hotkeys. Please use a new coldkey for the swap."
+            if raise_error:
+                raise ValueError(error_msg)
+            return ExtrinsicResponse(
+                success=False,
+                message=error_msg,
+                extrinsic_receipt=None,
+            )
+
         # Compute hash of new coldkey
         new_coldkey = Keypair(ss58_address=new_coldkey_ss58)
         new_coldkey_hash = compute_coldkey_hash(new_coldkey)

bittensor/core/extrinsics/coldkey_swap.py

@@ -60,6 +60,7 @@ def announce_coldkey_swap_extrinsic(
         - A swap cost is charged when making the first announcement (not when reannouncing).
         - After making an announcement, all transactions from the coldkey are blocked except for `swap_coldkey_announced`.
         - The swap can only be executed after the delay period has passed (check via `get_coldkey_swap_announcement`).
+        - The destination coldkey cannot have any staking hotkeys. It must be completely new without any staking activity.
         - See: <https://docs.learnbittensor.org/keys/coldkey-swap>
     """
     try:
@@ -68,6 +69,17 @@ def announce_coldkey_swap_extrinsic(
         ).success:
             return unlocked
 
+        staking_hotkeys = subtensor.get_staking_hotkeys(new_coldkey_ss58)
+        if staking_hotkeys:
+            error_msg = "Destination coldkey cannot have any staking hotkeys. Please use a new coldkey for the swap."
+            if raise_error:
+                raise ValueError(error_msg)
+            return ExtrinsicResponse(
+                success=False,
+                message=error_msg,
+                extrinsic_receipt=None,
+            )
+
         # Compute hash of new coldkey
         new_keypair = Keypair(
             ss58_address=new_coldkey_ss58,

bittensor/core/subtensor.py

@@ -3666,6 +3666,27 @@ def get_stake_weight(self, netuid: int, block: Optional[int] = None) -> list[flo
         )
         return [u16_normalized_float(w) for w in cast(list[int], result or [])]
 
+    def get_staking_hotkeys(
+        self, coldkey_ss58: str, block: Optional[int] = None
+    ) -> list[str]:
+        """
+        Retrieves the hotkeys that have staked for a given coldkey.
+
+        Parameters:
+            coldkey_ss58: The SS58 address of the coldkey.
+            block: The block number at which to query the stake information.
+
+        Returns:
+            A list of hotkey SS58 addresses that have staked for the given coldkey.
+        """
+        result = self.substrate.query(
+            module="SubtensorModule",
+            storage_function="StakingHotkeys",
+            params=[coldkey_ss58],
+            block_hash=self.determine_block_hash(block),
+        )
+        return [decode_account_id(hotkey[0]) for hotkey in result or []]
+
     def get_start_call_delay(self, block: Optional[int] = None) -> int:
         """
         Retrieves the start call delay in blocks.

bittensor/extras/subtensor_api/staking.py

@@ -31,6 +31,7 @@ def __init__(self, subtensor: Union["_Subtensor", "_AsyncSubtensor"]):
         self.get_stake_info_for_coldkeys = subtensor.get_stake_info_for_coldkeys
         self.get_stake_movement_fee = subtensor.get_stake_movement_fee
         self.get_stake_weight = subtensor.get_stake_weight
+        self.get_staking_hotkeys = subtensor.get_staking_hotkeys
         self.get_unstake_fee = subtensor.get_unstake_fee
         self.move_stake = subtensor.move_stake
         self.set_auto_stake = subtensor.set_auto_stake

tests/unit_tests/extrinsics/asyncex/test_coldkey_swap.py

@@ -22,6 +22,9 @@ async def test_announce_coldkey_swap_extrinsic(subtensor, mocker):
         "unlock_wallet",
         return_value=ExtrinsicResponse(success=True, message="Unlocked"),
     )
+    mocked_get_staking_hotkeys = mocker.patch.object(
+        subtensor, "get_staking_hotkeys", new=mocker.AsyncMock(return_value=[])
+    )
     mocked_keypair = mocker.patch(
         "bittensor.core.extrinsics.asyncex.coldkey_swap.Keypair"
     )
@@ -55,6 +58,7 @@ async def test_announce_coldkey_swap_extrinsic(subtensor, mocker):
 
     # Asserts
     mocked_unlock_wallet.assert_called_once_with(wallet, False)
+    mocked_get_staking_hotkeys.assert_awaited_once_with(new_coldkey_ss58)
     mocked_keypair.assert_called_once_with(ss58_address=new_coldkey_ss58)
     mocked_compute_hash.assert_called_once_with(mocked_keypair_instance)
     mocked_subtensor_module.assert_called_once_with(subtensor)
@@ -85,6 +89,9 @@ async def test_announce_coldkey_swap_extrinsic_with_mev_protection(subtensor, mo
         "unlock_wallet",
         return_value=ExtrinsicResponse(success=True, message="Unlocked"),
     )
+    mocked_get_staking_hotkeys = mocker.patch.object(
+        subtensor, "get_staking_hotkeys", new=mocker.AsyncMock(return_value=[])
+    )
     mocked_keypair = mocker.patch(
         "bittensor.core.extrinsics.asyncex.coldkey_swap.Keypair"
     )
@@ -118,6 +125,7 @@ async def test_announce_coldkey_swap_extrinsic_with_mev_protection(subtensor, mo
 
     # Asserts
     mocked_unlock_wallet.assert_called_once_with(wallet, False)
+    mocked_get_staking_hotkeys.assert_awaited_once_with(new_coldkey_ss58)
     mocked_subtensor_module.assert_called_once_with(subtensor)
     mocked_pallet_instance.announce_coldkey_swap.assert_awaited_once_with(
         new_coldkey_hash="0x" + "00" * 32

tests/unit_tests/extrinsics/test_coldkey_swap.py

@@ -20,6 +20,9 @@ def test_announce_coldkey_swap_extrinsic(subtensor, mocker):
         "unlock_wallet",
         return_value=ExtrinsicResponse(success=True, message="Unlocked"),
     )
+    mocked_get_staking_hotkeys = mocker.patch.object(
+        subtensor, "get_staking_hotkeys", return_value=[]
+    )
     mocked_keypair = mocker.patch("bittensor.core.extrinsics.coldkey_swap.Keypair")
     mocked_keypair_instance = mocker.MagicMock()
     mocked_keypair_instance.public_key = b"\x00" * 32
@@ -49,6 +52,7 @@ def test_announce_coldkey_swap_extrinsic(subtensor, mocker):
 
     # Asserts
     mocked_unlock_wallet.assert_called_once_with(wallet, False)
+    mocked_get_staking_hotkeys.assert_called_once_with(new_coldkey_ss58)
     mocked_keypair.assert_called_once_with(ss58_address=new_coldkey_ss58)
     mocked_compute_hash.assert_called_once_with(mocked_keypair_instance)
     mocked_subtensor_module.assert_called_once_with(subtensor)
@@ -78,6 +82,9 @@ def test_announce_coldkey_swap_extrinsic_with_mev_protection(subtensor, mocker):
         "unlock_wallet",
         return_value=ExtrinsicResponse(success=True, message="Unlocked"),
     )
+    mocked_get_staking_hotkeys = mocker.patch.object(
+        subtensor, "get_staking_hotkeys", return_value=[]
+    )
     mocked_keypair = mocker.patch("bittensor.core.extrinsics.coldkey_swap.Keypair")
     mocked_keypair_instance = mocker.MagicMock()
     mocked_keypair_instance.public_key = b"\x00" * 32
@@ -107,6 +114,7 @@ def test_announce_coldkey_swap_extrinsic_with_mev_protection(subtensor, mocker):
 
     # Asserts
     mocked_unlock_wallet.assert_called_once_with(wallet, False)
+    mocked_get_staking_hotkeys.assert_called_once_with(new_coldkey_ss58)
     mocked_subtensor_module.assert_called_once_with(subtensor)
     mocked_pallet_instance.announce_coldkey_swap.assert_called_once_with(
         new_coldkey_hash="0x" + "00" * 32