Feat/coldkey swap clear

bittensor_cli/cli.py

@@ -4144,7 +4144,11 @@ def wallet_swap_coldkey(
         self,
         action: str = typer.Argument(
             None,
-            help="Action to perform: 'announce' to announce intent, 'execute' to complete swap after delay, 'dispute' to freeze the swap.",
+            help=(
+                "Action to perform: 'announce' to announce intent, "
+                "'execute' to complete swap after delay, 'dispute' to freeze the swap, "
+                "'clear' to withdraw announcement, 'check' to view status."
+            ),
         ),
         wallet_name: Optional[str] = Options.wallet_name,
         wallet_path: Optional[str] = Options.wallet_path,
@@ -4176,6 +4180,9 @@ def wallet_swap_coldkey(
         If you suspect compromise, you can [bold]Dispute[/bold] an active announcement to freeze
         all activity for the coldkey until the triumvirate can intervene.
 
+        If you want to withdraw your announcement, you can [bold]Clear[/bold] (withdraw) an announcement once the
+        reannouncement delay has elapsed.
+
         EXAMPLES
 
         Step 1 - Announce your intent to swap:
@@ -4190,9 +4197,13 @@ def wallet_swap_coldkey(
 
         [green]$[/green] btcli wallet swap-coldkey dispute
 
-        Check status of pending swaps:
+        Clear (withdraw) an announcement:
+
+        [green]$[/green] btcli wallet swap-coldkey clear
 
-        [green]$[/green] btcli wallet swap-check
+        Check status of your swap:
+
+        [green]$[/green] btcli wallet swap-coldkey check
         """
         self.verbosity_handler(quiet, verbose, prompt=False, json_output=False)
 
@@ -4201,18 +4212,19 @@ def wallet_swap_coldkey(
                 "\n[bold][blue]Coldkey Swap Actions:[/blue][/bold]\n"
                 "  [dark_sea_green3]announce[/dark_sea_green3] - Start the swap process (pays fee, starts delay timer)\n"
                 "  [dark_sea_green3]execute[/dark_sea_green3]  - Complete the swap (after delay period)\n"
-                "  [dark_sea_green3]dispute[/dark_sea_green3]  - Freeze the swap process if you suspect compromise\n\n"
-                "  [dim]You can check the current status of your swap with 'btcli wallet swap-check'.[/dim]\n"
+                "  [dark_sea_green3]dispute[/dark_sea_green3]  - Freeze the swap process if you suspect compromise\n"
+                "  [dark_sea_green3]clear[/dark_sea_green3]    - Withdraw your swap announcement\n"
+                "  [dark_sea_green3]check[/dark_sea_green3]    - Check the status of your swap\n\n"
             )
             action = Prompt.ask(
                 "Select action",
-                choices=["announce", "execute", "dispute"],
+                choices=["announce", "execute", "dispute", "clear", "check"],
                 default="announce",
             )
 
-        if action.lower() not in ("announce", "execute", "dispute"):
+        if action.lower() not in ("announce", "execute", "dispute", "clear", "check"):
             print_error(
-                f"Invalid action: {action}. Must be 'announce', 'execute', or 'dispute'."
+                f"Invalid action: {action}. Must be 'announce', 'execute', 'dispute', 'clear', or 'check'."
             )
             raise typer.Exit(1)
 
@@ -4233,7 +4245,7 @@ def wallet_swap_coldkey(
         )
 
         new_wallet_coldkey_ss58 = None
-        if action != "dispute":
+        if action not in ("dispute", "clear", "check"):
             if not new_wallet_or_ss58:
                 new_wallet_or_ss58 = Prompt.ask(
                     "Enter the [blue]new wallet name[/blue] or [blue]SS58 address[/blue] of the new coldkey",
@@ -4285,6 +4297,24 @@ def wallet_swap_coldkey(
                     mev_protection=mev_protection,
                 )
             )
+        elif action == "clear":
+            return self._run_command(
+                wallets.clear_coldkey_swap_announcement(
+                    wallet=wallet,
+                    subtensor=self.initialize_chain(network),
+                    decline=decline,
+                    quiet=quiet,
+                    prompt=prompt,
+                    mev_protection=mev_protection,
+                )
+            )
+        elif action == "check":
+            return self._run_command(
+                wallets.check_swap_status(
+                    subtensor=self.initialize_chain(network),
+                    origin_ss58=wallet.coldkeypub.ss58_address,
+                )
+            )
         else:
             return self._run_command(
                 wallets.execute_coldkey_swap(

bittensor_cli/src/commands/wallets.py

@@ -2224,7 +2224,7 @@ async def announce_coldkey_swap(
         console.print(details_table)
         console.print(
             f"\n[dim]After the delay, run:"
-            f"\n[green]btcli wallet swap-coldkey execute --new-coldkey {new_coldkey_ss58}[/green]"
+            f"\n[green]btcli wallet swap-coldkey execute --new-coldkey {new_coldkey_ss58} --wallet-name {wallet.name}[/green]"
         )
 
     return True
@@ -2345,6 +2345,134 @@ async def dispute_coldkey_swap(
     return True
 
 
+async def clear_coldkey_swap_announcement(
+    wallet: Wallet,
+    subtensor: SubtensorInterface,
+    decline: bool = False,
+    quiet: bool = False,
+    prompt: bool = True,
+    mev_protection: bool = False,
+) -> bool:
+    """Clear (withdraw) a pending coldkey swap announcement.
+
+    The announcement can only be cleared after the reannouncement delay has elapsed
+    past the execution block, and the swap must not be disputed.
+
+    Args:
+        wallet: Wallet that owns the announcement (must be the announcing coldkey).
+        subtensor: Connection to the Bittensor network.
+        decline: If True, default to declining at confirmation prompt.
+        quiet: If True, skip confirmation prompts and proceed.
+        prompt: If True, show confirmation prompts.
+        mev_protection: If True, encrypt the extrinsic with MEV protection.
+
+    Returns:
+        bool: True if the clear extrinsic was included successfully, else False.
+    """
+    block_hash = await subtensor.substrate.get_chain_head()
+    announcement, dispute, current_block, reannounce_delay = await asyncio.gather(
+        subtensor.get_coldkey_swap_announcement(
+            wallet.coldkeypub.ss58_address, block_hash=block_hash
+        ),
+        subtensor.get_coldkey_swap_dispute(
+            wallet.coldkeypub.ss58_address, block_hash=block_hash
+        ),
+        subtensor.substrate.get_block_number(block_hash=block_hash),
+        subtensor.get_coldkey_swap_reannouncement_delay(block_hash=block_hash),
+    )
+
+    if not announcement:
+        print_error(
+            f"No coldkey swap announcement found for {wallet.coldkeypub.ss58_address}.\n"
+            "Nothing to clear."
+        )
+        return False
+
+    if dispute is not None:
+        console.print(
+            f"[yellow]Swap is disputed at block {dispute}.[/yellow] "
+            "Cannot clear a disputed announcement."
+        )
+        return False
+
+    clear_block = announcement.execution_block + reannounce_delay
+    if current_block < clear_block:
+        remaining = clear_block - current_block
+        console.print(
+            f"[yellow]Cannot clear yet.[/yellow] "
+            f"You can clear after block {clear_block} ({blocks_to_duration(remaining)} from now).\n"
+            f"Current block: {current_block}"
+        )
+        return False
+
+    info = create_key_value_table("Clear Coldkey Swap Announcement\n")
+    info.add_row(
+        "Coldkey", f"[{COLORS.G.CK}]{wallet.coldkeypub.ss58_address}[/{COLORS.G.CK}]"
+    )
+    info.add_row("Announced Hash", f"[dim]{announcement.new_coldkey_hash}[/dim]")
+    info.add_row("Execution Block", str(announcement.execution_block))
+    info.add_row(
+        "Status",
+        "[yellow]Pending[/yellow]"
+        if current_block < announcement.execution_block
+        else "[green]Ready[/green]",
+    )
+    console.print(info)
+
+    if prompt and not confirm_action(
+        "Proceed with clearing this swap announcement?",
+        decline=decline,
+        quiet=quiet,
+    ):
+        return False
+
+    if not unlock_key(wallet).success:
+        return False
+
+    with console.status(
+        ":satellite: Clearing coldkey swap announcement on-chain..."
+    ) as status:
+        call = await subtensor.substrate.compose_call(
+            call_module="SubtensorModule",
+            call_function="clear_coldkey_swap_announcement",
+            call_params={},
+        )
+        success, err_msg, ext_receipt = await subtensor.sign_and_send_extrinsic(
+            call,
+            wallet,
+            wait_for_inclusion=True,
+            wait_for_finalization=True,
+            mev_protection=mev_protection,
+        )
+
+        if not success:
+            print_error(f"Failed to clear coldkey swap announcement: {err_msg}")
+            return False
+
+        if mev_protection:
+            inner_hash = err_msg
+            mev_success, mev_error, ext_receipt = await wait_for_extrinsic_by_hash(
+                subtensor=subtensor,
+                extrinsic_hash=inner_hash,
+                submit_block_hash=ext_receipt.block_hash,
+                status=status,
+            )
+            if not mev_success:
+                print_error(
+                    f"Failed to clear coldkey swap announcement: {mev_error}",
+                    status=status,
+                )
+                return False
+
+        print_success("[dark_sea_green3]Coldkey swap announcement cleared.")
+        await print_extrinsic_id(ext_receipt)
+
+    console.print(
+        "\n[dim]Your coldkey is no longer locked by a pending swap announcement.[/dim]"
+    )
+    return True
+
+
 async def execute_coldkey_swap(
     wallet: Wallet,
     subtensor: SubtensorInterface,
@@ -2506,10 +2634,11 @@ async def check_swap_status(
     """
     block_hash = await subtensor.substrate.get_chain_head()
     if origin_ss58:
-        announcement, dispute, current_block = await asyncio.gather(
+        announcement, dispute, current_block, reannounce_delay = await asyncio.gather(
             subtensor.get_coldkey_swap_announcement(origin_ss58, block_hash=block_hash),
             subtensor.get_coldkey_swap_dispute(origin_ss58, block_hash=block_hash),
             subtensor.substrate.get_block_number(block_hash=block_hash),
+            subtensor.get_coldkey_swap_reannouncement_delay(block_hash=block_hash),
         )
         if not announcement:
             console.print(
@@ -2521,10 +2650,11 @@ async def check_swap_status(
         disputes = [(origin_ss58, dispute)] if dispute is not None else []
 
     else:
-        announcements, disputes, current_block = await asyncio.gather(
+        announcements, disputes, current_block, reannounce_delay = await asyncio.gather(
             subtensor.get_coldkey_swap_announcements(block_hash=block_hash),
             subtensor.get_coldkey_swap_disputes(block_hash=block_hash),
             subtensor.substrate.get_block_number(block_hash=block_hash),
+            subtensor.get_coldkey_swap_reannouncement_delay(block_hash=block_hash),
         )
         if not announcements:
             console.print(
@@ -2563,6 +2693,7 @@ async def check_swap_status(
         Column("Execution Block", justify="right", style="dark_sea_green3"),
         Column("Time Remaining", justify="right", style="yellow"),
         Column("Status", justify="center", style="green"),
+        Column("Clear Announcement", justify="right", style="yellow"),
         title=f"\n[{COLOR_PALETTE['GENERAL']['HEADER']}]Pending Coldkey Swap Announcements\nCurrent Block: {current_block}\n",
         show_header=True,
         show_edge=False,
@@ -2577,18 +2708,28 @@ async def check_swap_status(
     for announcement in announcements:
         dispute_block = dispute_map.get(announcement.coldkey)
         remaining_blocks = announcement.execution_block - current_block
+        clear_block = announcement.execution_block + reannounce_delay
+        clear_remaining = clear_block - current_block
         if dispute_block is not None:
             status = "[red]Disputed[/red]"
             time_str = f"Disputed @ {dispute_block}"
             status_label = "disputed"
+            clear_str = "[red]Disputed[/red]"
         elif remaining_blocks <= 0:
             status = "Ready"
             time_str = "[green]Ready[/green]"
             status_label = "ready"
+            if clear_remaining <= 0:
+                clear_str = "[green]Ready[/green]"
+            else:
+                clear_str = (
+                    f"Block {clear_block} ({blocks_to_duration(clear_remaining)})"
+                )
         else:
             status = "Pending"
             time_str = blocks_to_duration(remaining_blocks)
             status_label = "pending"
+            clear_str = f"Block {clear_block} ({blocks_to_duration(clear_remaining)})"
         hash_display = f"{announcement.new_coldkey_hash[:12]}...{announcement.new_coldkey_hash[-6:]}"
 
         table.add_row(
@@ -2597,6 +2738,7 @@ async def check_swap_status(
             str(announcement.execution_block),
             time_str,
             status,
+            clear_str,
         )
 
         payload["announcements"].append(
@@ -2607,6 +2749,8 @@ async def check_swap_status(
                 "status": status_label,
                 "time_remaining_blocks": max(0, remaining_blocks),
                 "disputed_block": dispute_block,
+                "clear_block": clear_block,
+                "clear_remaining_blocks": max(0, clear_remaining),
             }
         )
 
@@ -2617,5 +2761,7 @@ async def check_swap_status(
     console.print(table)
     console.print(
         "\n[dim]To execute a ready swap:[/dim] "
-        "[green]btcli wallet swap-coldkey execute[/green]"
+        "[green]btcli wallet swap-coldkey execute[/green]\n"
+        "[dim]To clear (withdraw) an announcement:[/dim] "
+        "[green]btcli wallet swap-coldkey clear[/green]"
     )