Feat/delegated claim type

bittensor_cli/cli.py

@@ -97,6 +97,7 @@
     add as add_stake,
     remove as remove_stake,
     claim as claim_stake,
+    validator_claim,
     wizard as stake_wizard,
 )
 from bittensor_cli.src.commands.subnets import (
@@ -163,7 +164,9 @@ def validate_claim_type(value: Optional[str]) -> Optional[claim_stake.ClaimType]
                 return member
         return claim_stake.ClaimType(value)
     except ValueError:
-        raise typer.BadParameter(f"'{value}' is not one of 'Keep', 'Swap'.")
+        raise typer.BadParameter(
+            f"'{value}' is not one of 'Keep', 'Swap', 'Delegated'."
+        )
 
 
 class Options:
@@ -1085,6 +1088,12 @@ def __init__(self):
         self.stake_app.command(
             "process-claim", rich_help_panel=HELP_PANELS["STAKE"]["CLAIM"]
         )(self.stake_process_claim)
+        self.stake_app.command(
+            "show-validator-claims", rich_help_panel=HELP_PANELS["STAKE"]["CLAIM"]
+        )(self.show_validator_claims)
+        self.stake_app.command(
+            "set-validator-claims", rich_help_panel=HELP_PANELS["STAKE"]["CLAIM"]
+        )(self.set_validator_claim_types)
 
         # stake-children commands
         children_app = typer.Typer()
@@ -1286,6 +1295,14 @@ def __init__(self):
             "unclaim",
             hidden=True,
         )(self.stake_set_claim_type)
+        self.stake_app.command(
+            "show-validator-claim",
+            hidden=True,
+        )(self.show_validator_claims)
+        self.stake_app.command(
+            "set-validator-claim",
+            hidden=True,
+        )(self.set_validator_claim_types)
 
         # Crowdloan
         self.app.add_typer(
@@ -5838,6 +5855,7 @@ def stake_set_claim_type(
         • [green]Swap[/green]: Future Root Alpha Emissions are swapped to TAO and added to root stake (default)
         • [yellow]Keep[/yellow]: Future Root Alpha Emissions are kept as Alpha tokens
         • [cyan]Keep Specific[/cyan]: Keep specific subnets as Alpha, swap others to TAO. You can use this type by selecting the netuids.
+        • [magenta]Delegated[/magenta]: Delegate claim choice to validator (inherits validator claim type)
 
         USAGE:
 
@@ -5846,6 +5864,7 @@ def stake_set_claim_type(
         [green]$[/green] btcli stake claim swap [cyan](Swap all subnets)[/cyan]
         [green]$[/green] btcli stake claim keep --netuids 1-5,10,20-30 [cyan](Keep specific subnets)[/cyan]
         [green]$[/green] btcli stake claim swap --netuids 1-30 [cyan](Swap specific subnets)[/cyan]
+        [green]$[/green] btcli stake claim delegated [cyan](Delegate claim choice to validator)[/cyan]
 
         With specific wallet:
 
@@ -5871,6 +5890,76 @@ def stake_set_claim_type(
             )
         )
 
+    def set_validator_claim_types(
+        self,
+        keep: Optional[str] = typer.Option(
+            None,
+            "--keep",
+            help="Subnets to keep emissions for (e.g. '1,3-5').",
+        ),
+        swap: Optional[str] = typer.Option(
+            None,
+            "--swap",
+            help="Subnets to swap emissions for (e.g. '2,6-10').",
+        ),
+        keep_all: bool = typer.Option(
+            False,
+            "--keep-all",
+            help="Set all registered subnets to Keep.",
+        ),
+        swap_all: bool = typer.Option(
+            False,
+            "--swap-all",
+            help="Set all registered subnets to Swap.",
+        ),
+        wallet_name: Optional[str] = Options.wallet_name,
+        wallet_path: Optional[str] = Options.wallet_path,
+        wallet_hotkey: Optional[str] = Options.wallet_hotkey,
+        network: Optional[list[str]] = Options.network,
+        proxy: Optional[str] = Options.proxy,
+        announce_only: bool = Options.announce_only,
+        prompt: bool = Options.prompt,
+        quiet: bool = Options.quiet,
+        verbose: bool = Options.verbose,
+    ):
+        """
+        Set the claim type for a validator across multiple subnets.
+
+        This command allows validators to specify how they want to handle emissions for each subnet:
+        - [bold]Keep[/bold]: Keep emissions as Alpha tokens on the subnet.
+        - [bold]Swap[/bold]: Automatically swap Alpha emissions to TAO on the root network.
+
+        You can set preferences for specific subnets using [blue]--keep[/blue] and [blue]--swap[/blue], or update all registered subnets using [blue]--keep-all[/blue] or [blue]--swap-all[/blue].
+
+        If no arguments are provided, an interactive editor will be launched.
+
+        EXAMPLES:
+        [green]$[/green] btcli stake set-validator-claim --keep 1,3-5 --swap 2,10
+        [green]$[/green] btcli stake set-validator-claim --keep-all
+        [green]$[/green] btcli stake set-validator-claim (Interactive mode)
+        """
+        self.verbosity_handler(quiet, verbose, False, prompt)
+        proxy = self.is_valid_proxy_name_or_ss58(proxy, announce_only)
+        wallet = self.wallet_ask(
+            wallet_name,
+            wallet_path,
+            wallet_hotkey,
+            ask_for=[WO.NAME, WO.HOTKEY],
+            validate=WV.WALLET_AND_HOTKEY,
+        )
+        return self._run_command(
+            validator_claim.set_validator_claim_type(
+                wallet=wallet,
+                subtensor=self.initialize_chain(network),
+                keep=keep,
+                swap=swap,
+                keep_all=keep_all,
+                swap_all=swap_all,
+                prompt=prompt,
+                proxy=proxy,
+            )
+        )
+
     def stake_process_claim(
         self,
         netuids: Optional[str] = Options.netuids,
@@ -5939,6 +6028,58 @@ def stake_process_claim(
             )
         )
 
+    def show_validator_claims(
+        self,
+        hotkey_ss58: Optional[str] = Options.wallet_hotkey_ss58,
+        wallet_name: Optional[str] = Options.wallet_name,
+        wallet_path: Optional[str] = Options.wallet_path,
+        network: Optional[list[str]] = Options.network,
+        prompt: bool = Options.prompt,
+        quiet: bool = Options.quiet,
+        verbose: bool = Options.verbose,
+        json_output: bool = Options.json_output,
+    ):
+        """
+        Show validator claim types (Keep/Swap) across all subnets for a validator hotkey.
+
+        EXAMPLES:
+        [green]$[/green] btcli stake show-validator-claims --hotkey 5Grw...
+        [green]$[/green] btcli stake show-validator-claims --wallet-name my_wallet --wallet-hotkey hk
+        """
+        self.verbosity_handler(quiet, verbose, json_output, False)
+
+        if not hotkey_ss58 and not wallet_name:
+            ss58_or_wallet = Prompt.ask(
+                "Enter the [blue]hotkey ss58 address[/blue] or [blue]wallet name[/blue]",
+                default=self.config.get("wallet_name") or defaults.wallet.name,
+            )
+            if is_valid_ss58_address(ss58_or_wallet):
+                hotkey_ss58 = ss58_or_wallet
+            else:
+                wallet_name = ss58_or_wallet
+
+        if hotkey_ss58:
+            if is_valid_ss58_address(hotkey_ss58):
+                vali_hk_ss58 = hotkey_ss58
+        else:
+            wallet = self.wallet_ask(
+                wallet_name,
+                wallet_path,
+                hotkey_ss58,
+                ask_for=[WO.NAME, WO.HOTKEY],
+                validate=WV.WALLET_AND_HOTKEY,
+            )
+            vali_hk_ss58 = get_hotkey_pub_ss58(wallet)
+
+        return self._run_command(
+            validator_claim.show_validator_claims(
+                subtensor=self.initialize_chain(network),
+                hotkey_ss58=vali_hk_ss58,
+                verbose=verbose,
+                json_output=json_output,
+            )
+        )
+
     def stake_get_children(
         self,
         wallet_name: Optional[str] = Options.wallet_name,
@@ -6023,9 +6164,9 @@ def stake_set_children(
         announce_only: bool = Options.announce_only,
         wait_for_inclusion: bool = Options.wait_for_inclusion,
         wait_for_finalization: bool = Options.wait_for_finalization,
+        prompt: bool = Options.prompt,
         quiet: bool = Options.quiet,
         verbose: bool = Options.verbose,
-        prompt: bool = Options.prompt,
         json_output: bool = Options.json_output,
     ):
         """

bittensor_cli/src/bittensor/subtensor_interface.py

@@ -2034,6 +2034,103 @@ async def get_all_coldkeys_claim_type(
 
         return root_claim_types
 
+    async def get_vali_claim_types_for_hk_netuid(
+        self,
+        hotkey_ss58: str,
+        netuid: int,
+        block_hash: Optional[str] = None,
+        reuse_block: bool = False,
+    ) -> dict:
+        """
+        Retrieves the validator claim type for a specific hotkey on a subnet.
+
+        Args:
+            hotkey_ss58: Validator hotkey SS58 address.
+            netuid: Subnet identifier.
+            block_hash: Optional block hash for the query.
+            reuse_block: Whether to reuse the last-used blockchain block hash.
+
+        Returns:
+            dict: Claim type information in one of these formats:
+                - {"type": "Swap"}
+                - {"type": "Keep"}
+        """
+        result = await self.query(
+            module="SubtensorModule",
+            storage_function="ValidatorClaimType",
+            params=[hotkey_ss58, netuid],
+            block_hash=block_hash,
+            reuse_block_hash=reuse_block,
+        )
+        claim_type_key = next(iter(result.keys()))
+        return {"type": claim_type_key}
+
+    async def get_all_vali_claim_types(
+        self,
+        block_hash: Optional[str] = None,
+        reuse_block: bool = False,
+    ) -> dict[int, str]:
+        """
+        Retrieves validator claim types for all netuids for all validator hotkeys.
+
+        Args:
+            block_hash: Optional block hash for the query.
+            reuse_block: Whether to reuse the last-used block hash.
+
+        Returns:
+            dict[int, str]: Mapping of netuid -> claim type ("Keep" or "Swap").
+        """
+        result = await self.substrate.query_map(
+            module="SubtensorModule",
+            storage_function="ValidatorClaimType",
+            params=[],
+            block_hash=block_hash,
+            reuse_block_hash=reuse_block,
+        )
+
+        claim_types: dict[str, dict[int, str]] = {}
+        async for hk_netuid, claim_type_data in result:
+            hotkey_ss58 = decode_account_id(hk_netuid[0])
+            netuid = int(hk_netuid[1])
+            claim_type_key = next(iter(claim_type_data.value.keys()))
+            if hotkey_ss58 not in claim_types:
+                claim_types[hotkey_ss58] = {}
+            claim_types[hotkey_ss58][netuid] = claim_type_key
+
+        return claim_types
+
+    async def get_vali_claim_types_for_hk(
+        self,
+        hotkey_ss58: str,
+        block_hash: Optional[str] = None,
+        reuse_block: bool = False,
+    ) -> dict[int, str]:
+        """
+        Retrieves validator claim types for all netuids for a given validator hotkey.
+
+        Args:
+            hotkey_ss58: Validator hotkey SS58 address.
+            block_hash: Optional block hash for the query.
+            reuse_block: Whether to reuse the last-used block hash.
+
+        Returns:
+            dict[int, str]: Mapping of netuid -> claim type ("Keep" or "Swap").
+        """
+        result = await self.substrate.query_map(
+            module="SubtensorModule",
+            storage_function="ValidatorClaimType",
+            params=[hotkey_ss58],
+            block_hash=block_hash,
+            reuse_block_hash=reuse_block,
+        )
+
+        claim_types: dict[int, str] = {}
+        async for netuid, claim_type_data in result:
+            claim_type_key = next(iter(claim_type_data.value.keys()))
+            claim_types[int(netuid)] = claim_type_key
+
+        return claim_types
+
     async def get_staking_hotkeys(
         self,
         coldkey_ss58: str,

bittensor_cli/src/commands/stake/claim.py

@@ -29,6 +29,7 @@
 class ClaimType(Enum):
     Keep = "Keep"
     Swap = "Swap"
+    Delegated = "Delegated"
 
 
 async def set_claim_type(
@@ -536,18 +537,22 @@ async def _ask_for_claim_types(
             "[yellow]Options:[/yellow]\n"
             "  • [green]Swap[/green] - Convert emissions to TAO\n"
             "  • [green]Keep[/green] - Keep emissions as Alpha\n"
-            "  • [green]Keep Specific[/green] - Keep selected subnets, swap others\n",
+            "  • [green]Keep Specific[/green] - Keep selected subnets, swap others\n"
+            "  • [green]Delegated[/green] - Delegate claim choice to validator\n",
         )
     )
 
     primary_choice = Prompt.ask(
         "\nSelect new root claim type",
-        choices=["keep", "swap", "cancel"],
+        choices=["keep", "swap", "delegated", "cancel"],
         default="cancel",
     )
     if primary_choice == "cancel":
         return None
 
+    if primary_choice == "delegated":
+        return {"type": "Delegated"}
+
     apply_to_all = Confirm.ask(
         f"\nSet {primary_choice.capitalize()} to ALL subnets?", default=True
     )
@@ -713,6 +718,10 @@ def _format_claim_type_display(
                 result += f"\n[yellow]  ⟳ Swap:[/yellow] {swap_display}"
 
         return result
+
+    elif claim_type == "Delegated":
+        return "[magenta]Delegated[/magenta]"
+
     else:
         return "[red]Unknown[/red]"
 
@@ -742,5 +751,7 @@ def _prepare_claim_type_args(claim_info: dict) -> dict:
     elif claim_type == "KeepSubnets":
         subnets = claim_info["subnets"]
         return {"KeepSubnets": {"subnets": subnets}}
+    elif claim_type == "Delegated":
+        return {"Delegated": None}
     else:
         raise ValueError(f"Unknown claim type: {claim_type}")