feat(cli,sdk): add support for "queue scripts" (#95)

fubuloubu · web-flow · commit cf06b2d8c285 · 2025-10-30T19:11:18.000Z
* refactor(MultiSend): encode data directly instead of using `txn` arg

* refactor(MultiSend): remove unreachable checks

* feat(SDK): add `.cli` module w/ `propose_batch` decorator

* feat(CLI): add `pending ensure` command to process "queue scripts"

* docs(SDK,Userguide): add docs &amp; userguide for `propose_from_simulation`

* fix(CLI): ensure full range of set

* fix(SDK): add additional check of nonce script name being numeric

* refactor(CLI): add prompt to `--submitter` option for queue script/`ensure`

* refactor(CLI): raise UsageError for errors in `pending ensure` cmd

* docs(SDK): add better docs for how to use args in queue scripts
diff --git a/ape_safe/_cli/pending.py b/ape_safe/_cli/pending.py
@@ -1,9 +1,10 @@
+import runpy
 from collections.abc import Sequence
 from typing import TYPE_CHECKING, Optional, Union, cast
 
 import click
 import rich
-from ape.cli import ConnectedProviderCommand
+from ape.cli import ConnectedProviderCommand, account_option
 from ape.exceptions import SignatureError
 from eth_typing import ChecksumAddress, Hash32
 from eth_utils import humanize_hash, to_hex
@@ -404,3 +405,80 @@ def _filter_tx_from_ids(
         return [x for x in txn_ids if x != txn.safe_tx_hash]
 
     return txn_ids
+
+
+@pending.command(cls=ConnectedProviderCommand)
+@safe_cli_ctx()
+@account_option("--submitter", prompt="Select an account to submit or propose transaction(s)")
+@safe_option
+def ensure(cli_ctx, ecosystem, network, submitter, safe):
+    """
+    Ensure Safe API queue matches `scripts/`
+
+    This command uses scripts from `scripts/<ecosystem-name>_<network-name>_nonce*.py`
+    (or `scripts/nonce*.py` if no network-specified scripts detected) and "replays" them from the
+    Safe's current on-chain nonce, up to the last script available in the folder. When executed
+    using a fork network, it will only perform a simulated validation of the script. When executed
+    using a live network, it will propose the transaction ONLY IF the transaction at that nonce
+    height does not match the expected calldata.
+
+    To enable this, scripts under `scripts/` must be properly named, and all use
+    `ape_safe.cli.propose_batch`.
+    """
+
+    if scripts_to_ensure := list(
+        cli_ctx.local_project.scripts_folder.glob(
+            f"{ecosystem.name}_{network.name.rstrip('-fork')}_nonce*.py"
+        )
+    ):
+        scripts_found_str = "\n".join(
+            str(fs.relative_to(cli_ctx.local_project.scripts_folder)) for fs in scripts_to_ensure
+        )
+        cli_ctx.logger.debug(f"Network-specific scripts found:\n{scripts_found_str}")
+
+    elif scripts_to_ensure := list(cli_ctx.local_project.scripts_folder.glob("nonce*.py")):
+        scripts_found_str = "\n".join(
+            str(fs.relative_to(cli_ctx.local_project.scripts_folder)) for fs in scripts_to_ensure
+        )
+        cli_ctx.logger.debug(f"Scripts found:\n{scripts_found_str}")
+
+    else:
+        raise click.UsageError("No queue scripts detected under `scripts/`.")
+
+    starting_nonce = safe.next_nonce
+    if not (
+        pending_scripts := {
+            nonce: script_path
+            for script_path in scripts_to_ensure
+            if "nonce" in script_path.stem
+            and (nonce := int(script_path.stem.split("nonce")[-1])) >= starting_nonce
+        }
+    ):
+        raise click.UsageError(
+            "No queue scripts detected under `scripts/` above current Safe nonce."
+        )
+
+    elif min(pending_scripts) != starting_nonce:
+        raise click.UsageError(
+            f"Next nonce for {safe.address} is {starting_nonce}, not {min(pending_scripts)}."
+        )
+
+    elif missing_nonces := sorted(
+        set(range(starting_nonce, max(pending_scripts) + 1)) - set(pending_scripts)
+    ):
+        display_str = ", ".join(map(str, missing_nonces))
+        raise click.UsageError(f"Missing queue scripts for nonce(s): {display_str}")
+
+    for nonce in sorted(pending_scripts):
+        script_path = pending_scripts[nonce].relative_to(cli_ctx.local_project.path)
+        if not (cmd := runpy.run_path(str(script_path), run_name=script_path.stem).get("cli")):
+            raise click.UsageError(f"No command `cli` detected in {script_path}.")
+
+        cli_ctx.logger.info(
+            f"Running queue script for nonce {nonce} ('{script_path}'):\n\n  {cmd.help}\n"
+        )
+        # NOTE: This matches signature from `ape_safe.cli:propose_batch`
+        cmd.callback.__wrapped__.func(cli_ctx, network, submitter, safe)
+
+    if not network.is_fork:
+        cli_ctx.logger.success(f"Queue for {safe.address} up-to-date!")
diff --git a/ape_safe/cli.py b/ape_safe/cli.py
@@ -0,0 +1,170 @@
+import inspect
+from typing import TYPE_CHECKING, Callable
+
+import click
+from ape.api.accounts import ImpersonatedAccount
+from ape.cli import (
+    ApeCliContextObject,
+    ConnectedProviderCommand,
+    account_option,
+    ape_cli_context,
+    network_option,
+)
+
+from ape_safe._cli.click_ext import safe_argument
+
+if TYPE_CHECKING:
+    from ape.api import AccountAPI, NetworkAPI
+
+    from ape_safe.accounts import SafeAccount
+
+
+def propose_from_simulation():
+    """
+    Create and propose a new SafeTx from transaction receipts inside a fork.
+
+    Usage::
+
+        @propose_from_simulation()
+        # NOTE: Name of decorated function *must* be called `cli`
+        # NOTE: Decorated function may have `safe` or `submitter` args in it
+        def cli():
+            # This entire function is executed within a fork/isolated context
+            # Use normal Ape features
+            my_contract = Contract("<address>")
+
+            # Any transaction is performed as if `sender=safe.address`
+            my_contract.mutableMethod(...)
+
+            # You can make assertions that will cause your simulation to fail if tripped
+            assert my_contract.viewMethod() == ...
+
+            # You can also add conditional calls
+            if my_contract.viewMethod() < some_number:
+                my_contract.mutableMethod()
+
+            # Once you are done with your transactions, the simulation will complete after exiting
+
+
+        # Once the simulation is complete, the decorator will collect all receipts it finds
+        # from the `safe` and collect them into a single SafeTx to propose to a public network
+    """
+
+    def decorator(
+        cmd: (
+            Callable[[], None]
+            | Callable[["SafeAccount"], None]
+            | Callable[["SafeAccount", "AccountAPI"], None]
+        ),
+    ) -> click.Command:
+        args: dict = {}
+        parameters = inspect.signature(cmd).parameters
+        if "safe" in parameters:
+            args["safe"] = None
+
+        if "submitter" in parameters:
+            args["submitter"] = None
+
+        if (script_name := cmd.__module__.split(".")[-1]).startswith("nonce"):
+            if not script_name[5:].isnumeric():
+                raise click.UsageError(
+                    f"Script 'scripts/{script_name}.py' must follow 'nonce<N>.py', "
+                    "where <N> is convertible to an integer value"
+                )
+
+            script_nonce = int(script_name[5:])
+
+        else:
+            script_nonce = None  # Auto-determine nonce later
+
+        @click.command(cls=ConnectedProviderCommand, name=cmd.__module__)
+        @ape_cli_context()
+        @network_option()
+        @account_option(
+            "--submitter",
+            prompt="Select an account to submit or propose transaction(s)",
+        )
+        @safe_argument
+        def new_cmd(
+            cli_ctx: ApeCliContextObject,
+            network: "NetworkAPI",
+            submitter: "AccountAPI",
+            safe: "SafeAccount",
+        ):
+            if "safe" in args:
+                args["safe"] = safe
+            if "submitter" in args:
+                args["submitter"] = submitter
+
+            # TODO: Use name of script to determine nonce? If starts with `nonce<XXX>.py`
+            batch = safe.create_batch()
+            total_gas_used = 0
+
+            with (
+                cli_ctx.chain_manager.isolate()
+                if network.is_fork
+                else cli_ctx.network_manager.fork()
+            ):
+                with cli_ctx.account_manager.use_sender(
+                    # NOTE: Use impersonated account to skip processing w/ `SafeAccount`
+                    ImpersonatedAccount(raw_address=safe.address)
+                ):
+                    # NOTE: Make sure the safe has money for gas
+                    if not safe.balance:
+                        safe.balance += int(1e20)
+
+                    cmd(*args.values())  # NOTE: 1 or more receipts should be mined from calling cmd
+
+                cli_ctx.logger.success(
+                    "Simulation complete, collecting receipts into batch to propose"
+                )
+
+                for txn in safe.history.sessional:
+                    if txn.sender != safe.address:
+                        raise RuntimeError("Don't execute other transactions!")
+
+                    elif txn.failed:
+                        raise RuntimeError("Transaction failed!")
+
+                    total_gas_used += txn.gas_used
+                    batch.add_from_receipt(txn)
+
+                # NOTE: After here, we are exiting the isolation context (either fork or rollback)
+
+            cli_ctx.logger.info(
+                f"Collected {len(batch.calls)} receipts into batch (gas used: {total_gas_used})"
+            )
+
+            if len(batch.calls) > 1:
+                safe_tx = batch.as_safe_tx(nonce=script_nonce)
+
+            else:  # When only one transaction receipt exits, just directly call that
+                cli_ctx.logger.info("Only 1 call found, calling directly instead of MultiSend")
+                txn = batch.calls[0]
+                safe_tx = safe.create_safe_tx(
+                    to=txn["target"], value=txn["value"], data=txn["callData"]
+                )
+
+            if network.is_fork:  # Testing, execute as a simulation (don't set nonce)
+                cli_ctx.logger.info("Using fork network, dry-running SafeTx")
+                safe.create_execute_transaction(safe_tx, {}, impersonate=True, submitter=submitter)
+
+            elif not (confirmations := safe.get_api_confirmations(safe_tx)):
+                # Real mainnet, propose if not already in queue
+                cli_ctx.logger.info("Using public network, proposing SafeTx to Safe API")
+                safe.propose_safe_tx(safe_tx, submitter=submitter)
+
+                safe_tx_id = safe_tx._message_hash_
+                cli_ctx.logger.success("'{safe_tx_id.to_0x_hex()}' proposed to queue")
+
+            else:
+                safe_tx_id = safe_tx._message_hash_
+                cli_ctx.logger.info(
+                    f"SafeTxID '{safe_tx_id.to_0x_hex()}' already in queue "
+                    f"({len(confirmations)}/{safe.confirmations_required})"
+                )
+
+        new_cmd.help = cmd.__doc__
+        return new_cmd
+
+    return decorator
diff --git a/ape_safe/multisend.py b/ape_safe/multisend.py
@@ -209,11 +209,9 @@ def as_safe_tx(self, safe: Optional[SafeAccount] = None, **safe_tx_kwargs) -> "S
         if not (safe or (safe := self.safe)):
             raise ValueError("Must provide `safe=` to call this function")
 
-        elif not isinstance(safe, SafeAccount):
-            raise ValueError("`safe=` must be a SafeAccount to use Multisend")
-
         return safe.create_safe_tx(
-            self.handler.as_transaction(b"".join(self.encoded_calls)),
+            to=self.contract.address,
+            data=self.handler.encode_input(b"".join(self.encoded_calls)),
             operation=OperationType.DELEGATECALL,
             **safe_tx_kwargs,
         )
@@ -226,9 +224,6 @@ def propose(
         if not (safe or (safe := self.safe)):
             raise ValueError("Must provide `safe=` to call this function")
 
-        elif not isinstance(safe, SafeAccount):
-            raise ValueError("`safe=` must be a SafeAccount to use Multisend")
-
         submitter = safe_tx_kwargs.pop("submitter", None)
         sigs_by_signer = safe_tx_kwargs.pop("sigs_by_signer", None)
         safe_tx = self.as_safe_tx(safe, **safe_tx_kwargs)
@@ -262,9 +257,6 @@ def as_transaction(
         if not (safe or (safe := self.safe)):
             raise ValueError("Must provide `safe=` to call this function")
 
-        elif not isinstance(safe, SafeAccount):
-            raise ValueError("`safe=` must be a SafeAccount to use Multisend")
-
         safe_tx_kwargs = {}
         for field in safe.safe_tx_def.__annotations__:
             if field in txn_kwargs:
diff --git a/docs/index.rst b/docs/index.rst
@@ -5,6 +5,7 @@
         - safe_management
         - multisend
         - modules
+        - production
     :commands:
         - mgmt
         - pending
diff --git a/docs/methoddocs/cli.md b/docs/methoddocs/cli.md
@@ -0,0 +1,4 @@
+```{eval-rst}
+.. automodule:: ape_safe.cli
+    :members:
+```
diff --git a/docs/userguides/production.md b/docs/userguides/production.md
diff --git a/setup.py b/setup.py
