tumbler.cli

tumbler.cli

Standalone command-line interface for the JoinMarket tumbler.

Mirrors the patterns used by :mod:taker.cli and :mod:maker.cli: configuration is loaded from (in priority order) CLI arguments, environment variables, the config file at ~/.joinmarket-ng/config.toml (or $JOINMARKET_DATA_DIR/config.toml), and built-in defaults.

The CLI is a thin wrapper around :mod:tumbler.builder, :mod:tumbler.persistence, and :mod:tumbler.runner. Plans are persisted to <data_dir>/schedules/<wallet_name>.yaml so the same file is used whether the tumble runs from the CLI or from jmwalletd.

Attributes

app = typer.Typer(name='jm-tumbler', help='JoinMarket tumbler - role-mixed CoinJoin schedules with YAML-persisted state', no_args_is_help=True) module-attribute

Classes

RunnerPacing

Bases: NamedTuple

Resolved inter-phase pacing knobs for :class:tumbler.runner.RunnerContext.

Source code in tumbler/src/tumbler/cli.py

class RunnerPacing(NamedTuple):
    """Resolved inter-phase pacing knobs for :class:`tumbler.runner.RunnerContext`."""

    min_confirmations_between_phases: int
    confirmation_poll_interval: float
    retry_delay_seconds: float

Attributes

confirmation_poll_interval: float instance-attribute

min_confirmations_between_phases: int instance-attribute

retry_delay_seconds: float instance-attribute

Functions:

config_init(data_dir: Annotated[Path | None, typer.Option('--data-dir', '-d', envvar='JOINMARKET_DATA_DIR', help='Data directory for JoinMarket files')] = None, config_file: Annotated[Path | None, typer.Option('--config-file', envvar='JOINMARKET_CONFIG_FILE', help='Config file path (decoupled from data dir). Defaults to <data-dir>/config.toml')] = None) -> None

Initialize the config file with default settings.

Source code in tumbler/src/tumbler/cli.py

@app.command("config-init")
def config_init(
    data_dir: Annotated[
        Path | None,
        typer.Option(
            "--data-dir",
            "-d",
            envvar="JOINMARKET_DATA_DIR",
            help="Data directory for JoinMarket files",
        ),
    ] = None,
    config_file: Annotated[
        Path | None,
        typer.Option(
            "--config-file",
            envvar="JOINMARKET_CONFIG_FILE",
            help="Config file path (decoupled from data dir). Defaults to <data-dir>/config.toml",
        ),
    ] = None,
) -> None:
    """Initialize the config file with default settings."""
    from jmcore.paths import get_default_data_dir

    if data_dir is None:
        data_dir = get_default_data_dir()
    config_path = ensure_config_file(data_dir, config_file=config_file)
    typer.echo(f"Config file created at: {config_path}")

delete_command(wallet_name: Annotated[str | None, typer.Option('--wallet-name', '-w', help='Wallet identifier; defaults to the mnemonic fingerprint')] = None, mnemonic_file: Annotated[Path | None, typer.Option('--mnemonic-file', '-f', help='Path to mnemonic file')] = None, prompt_bip39_passphrase: Annotated[bool, typer.Option('--prompt-bip39-passphrase', help='Prompt for BIP39 passphrase interactively')] = False, yes: Annotated[bool, typer.Option('--yes', '-y', help='Skip confirmation prompt')] = False, data_dir: Annotated[Path | None, typer.Option('--data-dir', envvar='JOINMARKET_DATA_DIR', help='Data directory (default: ~/.joinmarket-ng or $JOINMARKET_DATA_DIR)')] = None, config_file: Annotated[Path | None, typer.Option('--config-file', envvar='JOINMARKET_CONFIG_FILE', help='Config file path (decoupled from data dir). Defaults to <data-dir>/config.toml')] = None, log_level: Annotated[str | None, typer.Option('--log-level', '-l')] = None) -> None

Delete the on-disk plan for wallet_name.

Source code in tumbler/src/tumbler/cli.py

@app.command("delete")
def delete_command(
    wallet_name: Annotated[
        str | None,
        typer.Option(
            "--wallet-name",
            "-w",
            help="Wallet identifier; defaults to the mnemonic fingerprint",
        ),
    ] = None,
    mnemonic_file: Annotated[
        Path | None, typer.Option("--mnemonic-file", "-f", help="Path to mnemonic file")
    ] = None,
    prompt_bip39_passphrase: Annotated[
        bool,
        typer.Option(
            "--prompt-bip39-passphrase",
            help="Prompt for BIP39 passphrase interactively",
        ),
    ] = False,
    yes: Annotated[bool, typer.Option("--yes", "-y", help="Skip confirmation prompt")] = False,
    data_dir: Annotated[
        Path | None,
        typer.Option(
            "--data-dir",
            envvar="JOINMARKET_DATA_DIR",
            help="Data directory (default: ~/.joinmarket-ng or $JOINMARKET_DATA_DIR)",
        ),
    ] = None,
    config_file: Annotated[
        Path | None,
        typer.Option(
            "--config-file",
            envvar="JOINMARKET_CONFIG_FILE",
            help="Config file path (decoupled from data dir). Defaults to <data-dir>/config.toml",
        ),
    ] = None,
    log_level: Annotated[str | None, typer.Option("--log-level", "-l")] = None,
) -> None:
    """Delete the on-disk plan for ``wallet_name``."""
    settings = setup_cli(log_level, data_dir=data_dir, config_file=config_file)
    data_dir = settings.get_data_dir()
    effective_wallet = _resolve_wallet_name(
        settings, wallet_name, mnemonic_file, prompt_bip39_passphrase
    )
    plan = _load_or_error(effective_wallet, data_dir)
    if plan.status == PlanStatus.RUNNING:
        logger.error("Plan is RUNNING; stop it before deleting.")
        raise typer.Exit(1)
    if not yes and not typer.confirm(
        f"Delete tumbler plan for {effective_wallet} (status={plan.status.value})?"
    ):
        typer.echo("Cancelled.")
        return
    if delete_plan_on_disk(effective_wallet, data_dir):
        typer.echo(f"Deleted {plan_path(effective_wallet, data_dir)}")
    else:
        typer.echo("Nothing to delete.")

main() -> None

Entry point.

Source code in tumbler/src/tumbler/cli.py

def main() -> None:
    """Entry point."""
    app()

plan_command(destinations: Annotated[list[str], typer.Option('--destination', '-d', help='External destination address (repeatable)')], mnemonic_file: Annotated[Path | None, typer.Option('--mnemonic-file', '-f', help='Path to mnemonic file')] = None, prompt_bip39_passphrase: Annotated[bool, typer.Option('--prompt-bip39-passphrase', help='Prompt for BIP39 passphrase interactively')] = False, wallet_name: Annotated[str | None, typer.Option('--wallet-name', '-w', help='Wallet identifier for the plan file; defaults to the mnemonic fingerprint')] = None, network: Annotated[NetworkType | None, typer.Option('--network', case_sensitive=False, help='Bitcoin network')] = None, backend_type: Annotated[str | None, typer.Option('--backend', '-b', help='Backend type: descriptor_wallet | neutrino')] = None, rpc_url: Annotated[str | None, typer.Option('--rpc-url', envvar='BITCOIN_RPC_URL', help='Bitcoin full node RPC URL')] = None, neutrino_url: Annotated[str | None, typer.Option('--neutrino-url', envvar='NEUTRINO_URL', help='Neutrino REST API URL')] = None, force: Annotated[bool, typer.Option('--force', help='Overwrite an existing pending plan')] = False, seed: Annotated[int | None, typer.Option('--seed', help='Seed the plan builder RNG for reproducible schedules')] = None, maker_count_min: Annotated[int | None, typer.Option(help='Minimum counterparty count per CJ; defaults to settings.taker.counterparty_count')] = None, maker_count_max: Annotated[int | None, typer.Option(help='Maximum counterparty count per CJ; defaults to settings.taker.counterparty_count')] = None, mincjamount_sats: Annotated[int, typer.Option(help='Minimum CJ amount in sats')] = 100000, include_maker_sessions: Annotated[bool, typer.Option('--maker-sessions/--no-maker-sessions')] = True, allow_few_destinations: Annotated[bool, typer.Option('--allow-few-destinations', help=f'Override the recommended minimum of {MIN_DESTINATIONS} destinations. Intended for development and automated testing only: fewer destinations expose users to pairwise re-aggregation heuristics.')] = False, data_dir: Annotated[Path | None, typer.Option('--data-dir', envvar='JOINMARKET_DATA_DIR', help='Data directory (default: ~/.joinmarket-ng or $JOINMARKET_DATA_DIR)')] = None, config_file: Annotated[Path | None, typer.Option('--config-file', envvar='JOINMARKET_CONFIG_FILE', help='Config file path (decoupled from data dir). Defaults to <data-dir>/config.toml')] = None, log_level: Annotated[str | None, typer.Option('--log-level', '-l')] = None) -> None

Build a tumbler plan for the given destinations and persist it.

Source code in tumbler/src/tumbler/cli.py

@app.command("plan")
def plan_command(
    destinations: Annotated[
        list[str],
        typer.Option("--destination", "-d", help="External destination address (repeatable)"),
    ],
    mnemonic_file: Annotated[
        Path | None, typer.Option("--mnemonic-file", "-f", help="Path to mnemonic file")
    ] = None,
    prompt_bip39_passphrase: Annotated[
        bool,
        typer.Option(
            "--prompt-bip39-passphrase",
            help="Prompt for BIP39 passphrase interactively",
        ),
    ] = False,
    wallet_name: Annotated[
        str | None,
        typer.Option(
            "--wallet-name",
            "-w",
            help="Wallet identifier for the plan file; defaults to the mnemonic fingerprint",
        ),
    ] = None,
    network: Annotated[
        NetworkType | None,
        typer.Option("--network", case_sensitive=False, help="Bitcoin network"),
    ] = None,
    backend_type: Annotated[
        str | None,
        typer.Option("--backend", "-b", help="Backend type: descriptor_wallet | neutrino"),
    ] = None,
    rpc_url: Annotated[
        str | None,
        typer.Option("--rpc-url", envvar="BITCOIN_RPC_URL", help="Bitcoin full node RPC URL"),
    ] = None,
    neutrino_url: Annotated[
        str | None,
        typer.Option("--neutrino-url", envvar="NEUTRINO_URL", help="Neutrino REST API URL"),
    ] = None,
    force: Annotated[
        bool,
        typer.Option(
            "--force",
            help="Overwrite an existing pending plan",
        ),
    ] = False,
    seed: Annotated[
        int | None,
        typer.Option("--seed", help="Seed the plan builder RNG for reproducible schedules"),
    ] = None,
    maker_count_min: Annotated[
        int | None,
        typer.Option(
            help=(
                "Minimum counterparty count per CJ; defaults to settings.taker.counterparty_count"
            ),
        ),
    ] = None,
    maker_count_max: Annotated[
        int | None,
        typer.Option(
            help=(
                "Maximum counterparty count per CJ; defaults to settings.taker.counterparty_count"
            ),
        ),
    ] = None,
    mincjamount_sats: Annotated[int, typer.Option(help="Minimum CJ amount in sats")] = 100_000,
    include_maker_sessions: Annotated[
        bool, typer.Option("--maker-sessions/--no-maker-sessions")
    ] = True,
    allow_few_destinations: Annotated[
        bool,
        typer.Option(
            "--allow-few-destinations",
            help=(
                "Override the recommended minimum of "
                f"{MIN_DESTINATIONS} destinations. Intended for development and "
                "automated testing only: fewer destinations expose users to "
                "pairwise re-aggregation heuristics."
            ),
        ),
    ] = False,
    data_dir: Annotated[
        Path | None,
        typer.Option(
            "--data-dir",
            envvar="JOINMARKET_DATA_DIR",
            help="Data directory (default: ~/.joinmarket-ng or $JOINMARKET_DATA_DIR)",
        ),
    ] = None,
    config_file: Annotated[
        Path | None,
        typer.Option(
            "--config-file",
            envvar="JOINMARKET_CONFIG_FILE",
            help="Config file path (decoupled from data dir). Defaults to <data-dir>/config.toml",
        ),
    ] = None,
    log_level: Annotated[str | None, typer.Option("--log-level", "-l")] = None,
) -> None:
    """Build a tumbler plan for the given destinations and persist it."""
    if len(destinations) < MIN_DESTINATIONS and not allow_few_destinations:
        logger.error(
            "at least {} destination addresses are recommended (got {}). "
            "Pass --allow-few-destinations to override.",
            MIN_DESTINATIONS,
            len(destinations),
        )
        raise typer.Exit(1)
    settings = setup_cli(log_level, data_dir=data_dir, config_file=config_file)
    ensure_config_file(settings.get_data_dir())
    data_dir = settings.get_data_dir()

    try:
        resolved = resolve_mnemonic(
            settings,
            mnemonic_file=mnemonic_file,
            prompt_bip39_passphrase=prompt_bip39_passphrase,
        )
    except (ValueError, FileNotFoundError) as exc:
        logger.error(str(exc))
        raise typer.Exit(1)
    if resolved is None:
        logger.error("Could not resolve a mnemonic; supply --mnemonic-file or configure one.")
        raise typer.Exit(1)

    effective_wallet = wallet_name or _wallet_name_from_mnemonic(
        resolved.mnemonic, resolved.bip39_passphrase or "", settings.network_config.network
    )

    existing: Plan | None
    try:
        existing = load_plan(effective_wallet, data_dir)
    except PlanNotFoundError:
        existing = None
    except PlanCorruptError as exc:
        logger.error(f"Tumbler plan is corrupt: {exc}")
        raise typer.Exit(1)

    if existing is not None and existing.status == PlanStatus.RUNNING:
        logger.error(
            f"A plan is already RUNNING for {effective_wallet}; use 'jm-tumbler stop' first."
        )
        raise typer.Exit(1)
    if existing is not None and existing.status == PlanStatus.PENDING and not force:
        logger.error("A pending plan already exists; pass --force to overwrite.")
        raise typer.Exit(1)

    try:
        balances, fee_rate, fee_rate_source = asyncio.run(
            _balances_for_mnemonic(
                settings=settings,
                mnemonic=resolved.mnemonic,
                passphrase=resolved.bip39_passphrase or "",
                network=network,
                backend_type=backend_type,
                rpc_url=rpc_url,
                neutrino_url=neutrino_url,
            )
        )
    except RuntimeError as exc:
        logger.error(str(exc))
        raise typer.Exit(1)

    if not any(v > 0 for v in balances.values()):
        logger.error("Wallet has no confirmed coins to tumble.")
        raise typer.Exit(1)

    try:
        effective_min = (
            maker_count_min if maker_count_min is not None else settings.taker.counterparty_count
        )
        effective_max = (
            maker_count_max if maker_count_max is not None else settings.taker.counterparty_count
        )
        params = TumbleParameters(
            destinations=list(destinations),
            mixdepth_balances=balances,
            maker_count_min=effective_min,
            maker_count_max=effective_max,
            mincjamount_sats=mincjamount_sats,
            include_maker_sessions=include_maker_sessions,
            seed=seed,
        )
        plan = PlanBuilder(wallet_name=effective_wallet, params=params).build()
    except ValueError as exc:
        logger.error(str(exc))
        raise typer.Exit(1)

    path = save_plan(plan, data_dir)
    typer.echo(f"Plan written to {path}")

    estimate = estimate_plan_costs(
        plan,
        mixdepth_balances=balances,
        max_cj_fee_abs_sats=settings.taker.max_cj_fee_abs,
        max_cj_fee_rel=settings.taker.max_cj_fee_rel,
        fee_rate_sat_vb=fee_rate,
        fee_rate_source=fee_rate_source,
        confirmation_block_count=settings.tumbler.min_confirmations_between_phases,
    )
    _summarise_plan(plan, estimate=estimate)

    # Echo the relevant taker config so the user knows what bounds were
    # applied -- these are the same knobs that gate every CJ during run.
    typer.echo("")
    typer.echo("Active taker config")
    typer.echo("-------------------")
    typer.echo(f"  max_cj_fee_abs:         {settings.taker.max_cj_fee_abs} sats")
    typer.echo(f"  max_cj_fee_rel:         {settings.taker.max_cj_fee_rel}")
    typer.echo(
        f"  counterparty_count:     {settings.taker.counterparty_count}"
        f" (plan range: {effective_min}-{effective_max})"
    )
    if settings.taker.fee_rate is not None:
        typer.echo(f"  fee_rate:               {settings.taker.fee_rate} sat/vB")
    elif settings.taker.fee_block_target is not None:
        typer.echo(f"  fee_block_target:       {settings.taker.fee_block_target} blocks")

resolve_runner_pacing(settings: Any, min_confirmations_override: int | None) -> RunnerPacing

Resolve runner pacing from [tumbler] settings with a CLI override.

--min-confirmations (when given) wins over the configured min_confirmations_between_phases. The confirmation poll interval and retry delay always come from settings ([tumbler] config section or TUMBLER__* env vars); previously they were silently ignored by the standalone CLI and only honored by jmwalletd.

Source code in tumbler/src/tumbler/cli.py

def resolve_runner_pacing(settings: Any, min_confirmations_override: int | None) -> RunnerPacing:
    """Resolve runner pacing from ``[tumbler]`` settings with a CLI override.

    ``--min-confirmations`` (when given) wins over the configured
    ``min_confirmations_between_phases``. The confirmation poll interval and
    retry delay always come from settings (``[tumbler]`` config section or
    ``TUMBLER__*`` env vars); previously they were silently ignored by the
    standalone CLI and only honored by jmwalletd.
    """
    tumbler_settings = settings.tumbler
    min_confirmations = (
        min_confirmations_override
        if min_confirmations_override is not None
        else tumbler_settings.min_confirmations_between_phases
    )
    return RunnerPacing(
        min_confirmations_between_phases=min_confirmations,
        confirmation_poll_interval=tumbler_settings.confirmation_poll_interval,
        retry_delay_seconds=tumbler_settings.retry_delay_seconds,
    )

run_command(mnemonic_file: Annotated[Path | None, typer.Option('--mnemonic-file', '-f', help='Path to mnemonic file')] = None, prompt_bip39_passphrase: Annotated[bool, typer.Option('--prompt-bip39-passphrase', help='Prompt for BIP39 passphrase interactively')] = False, wallet_name: Annotated[str | None, typer.Option('--wallet-name', '-w', help='Wallet identifier; defaults to the mnemonic fingerprint')] = None, network: Annotated[NetworkType | None, typer.Option('--network', case_sensitive=False)] = None, backend_type: Annotated[str | None, typer.Option('--backend', '-b')] = None, rpc_url: Annotated[str | None, typer.Option('--rpc-url', envvar='BITCOIN_RPC_URL')] = None, neutrino_url: Annotated[str | None, typer.Option('--neutrino-url', envvar='NEUTRINO_URL')] = None, directory_servers: Annotated[str | None, typer.Option('--directory', '-D', envvar='DIRECTORY_SERVERS')] = None, tor_socks_host: Annotated[str | None, typer.Option(help='Tor SOCKS host override')] = None, tor_socks_port: Annotated[int | None, typer.Option(help='Tor SOCKS port override')] = None, fee_rate: Annotated[float | None, typer.Option('--fee-rate', help='Manual fee rate in sat/vB (mutually exclusive with --block-target). Required when the backend is neutrino.')] = None, block_target: Annotated[int | None, typer.Option('--block-target', help='Target blocks for fee estimation (mutually exclusive with --fee-rate). Not supported with the neutrino backend.')] = None, min_confirmations_between_phases: Annotated[int | None, typer.Option('--min-confirmations', help='Confirmations required before the next phase starts (0 disables gating). Defaults to the tumbler.min_confirmations_between_phases setting (6).')] = None, counterparties: Annotated[int | None, typer.Option('--counterparties', min=1, max=20, help='Override the counterparty count for every phase at runtime. Useful when the configured count is unavailable on the chosen network.')] = None, resume: Annotated[bool, typer.Option('--resume', help='Resume a plan that ended in a terminal state (FAILED, CANCELLED, or stuck-RUNNING). Completed phases are kept; all other phases are reset to PENDING and the runner picks up at the first non-completed phase. Has no effect on a plan that is already PENDING.')] = False, data_dir: Annotated[Path | None, typer.Option('--data-dir', envvar='JOINMARKET_DATA_DIR', help='Data directory (default: ~/.joinmarket-ng or $JOINMARKET_DATA_DIR)')] = None, config_file: Annotated[Path | None, typer.Option('--config-file', envvar='JOINMARKET_CONFIG_FILE', help='Config file path (decoupled from data dir). Defaults to <data-dir>/config.toml')] = None, log_level: Annotated[str | None, typer.Option('--log-level', '-l')] = None) -> None

Execute the saved plan for a wallet to completion.

Source code in tumbler/src/tumbler/cli.py

@app.command("run")
def run_command(
    mnemonic_file: Annotated[
        Path | None, typer.Option("--mnemonic-file", "-f", help="Path to mnemonic file")
    ] = None,
    prompt_bip39_passphrase: Annotated[
        bool,
        typer.Option("--prompt-bip39-passphrase", help="Prompt for BIP39 passphrase interactively"),
    ] = False,
    wallet_name: Annotated[
        str | None,
        typer.Option(
            "--wallet-name", "-w", help="Wallet identifier; defaults to the mnemonic fingerprint"
        ),
    ] = None,
    network: Annotated[NetworkType | None, typer.Option("--network", case_sensitive=False)] = None,
    backend_type: Annotated[
        str | None,
        typer.Option("--backend", "-b"),
    ] = None,
    rpc_url: Annotated[str | None, typer.Option("--rpc-url", envvar="BITCOIN_RPC_URL")] = None,
    neutrino_url: Annotated[
        str | None, typer.Option("--neutrino-url", envvar="NEUTRINO_URL")
    ] = None,
    directory_servers: Annotated[
        str | None,
        typer.Option("--directory", "-D", envvar="DIRECTORY_SERVERS"),
    ] = None,
    tor_socks_host: Annotated[str | None, typer.Option(help="Tor SOCKS host override")] = None,
    tor_socks_port: Annotated[int | None, typer.Option(help="Tor SOCKS port override")] = None,
    fee_rate: Annotated[
        float | None,
        typer.Option(
            "--fee-rate",
            help=(
                "Manual fee rate in sat/vB (mutually exclusive with --block-target). "
                "Required when the backend is neutrino."
            ),
        ),
    ] = None,
    block_target: Annotated[
        int | None,
        typer.Option(
            "--block-target",
            help=(
                "Target blocks for fee estimation (mutually exclusive with --fee-rate). "
                "Not supported with the neutrino backend."
            ),
        ),
    ] = None,
    min_confirmations_between_phases: Annotated[
        int | None,
        typer.Option(
            "--min-confirmations",
            help=(
                "Confirmations required before the next phase starts (0 disables "
                "gating). Defaults to the tumbler.min_confirmations_between_phases "
                "setting (6)."
            ),
        ),
    ] = None,
    counterparties: Annotated[
        int | None,
        typer.Option(
            "--counterparties",
            min=1,
            max=20,
            help=(
                "Override the counterparty count for every phase at runtime. "
                "Useful when the configured count is unavailable on the chosen network."
            ),
        ),
    ] = None,
    resume: Annotated[
        bool,
        typer.Option(
            "--resume",
            help=(
                "Resume a plan that ended in a terminal state (FAILED, "
                "CANCELLED, or stuck-RUNNING). Completed phases are kept; "
                "all other phases are reset to PENDING and the runner picks "
                "up at the first non-completed phase. Has no effect on a "
                "plan that is already PENDING."
            ),
        ),
    ] = False,
    data_dir: Annotated[
        Path | None,
        typer.Option(
            "--data-dir",
            envvar="JOINMARKET_DATA_DIR",
            help="Data directory (default: ~/.joinmarket-ng or $JOINMARKET_DATA_DIR)",
        ),
    ] = None,
    config_file: Annotated[
        Path | None,
        typer.Option(
            "--config-file",
            envvar="JOINMARKET_CONFIG_FILE",
            help="Config file path (decoupled from data dir). Defaults to <data-dir>/config.toml",
        ),
    ] = None,
    log_level: Annotated[str | None, typer.Option("--log-level", "-l")] = None,
) -> None:
    """Execute the saved plan for a wallet to completion."""
    settings = setup_cli(log_level, data_dir=data_dir, config_file=config_file)
    ensure_config_file(settings.get_data_dir())
    data_dir = settings.get_data_dir()

    if fee_rate is not None and block_target is not None:
        logger.error("--fee-rate and --block-target are mutually exclusive.")
        raise typer.Exit(1)

    effective_backend_type = backend_type or settings.bitcoin.backend_type
    if effective_backend_type == "neutrino" and fee_rate is None:
        logger.error("Neutrino backend cannot estimate fees; pass --fee-rate <sat/vB> to proceed.")
        raise typer.Exit(1)

    try:
        resolved = resolve_mnemonic(
            settings,
            mnemonic_file=mnemonic_file,
            prompt_bip39_passphrase=prompt_bip39_passphrase,
        )
    except (ValueError, FileNotFoundError) as exc:
        logger.error(str(exc))
        raise typer.Exit(1)
    if resolved is None:
        logger.error("Could not resolve a mnemonic.")
        raise typer.Exit(1)

    effective_wallet = wallet_name or _wallet_name_from_mnemonic(
        resolved.mnemonic, resolved.bip39_passphrase or "", settings.network_config.network
    )
    plan = _load_or_error(effective_wallet, data_dir)
    if plan.status in (PlanStatus.COMPLETED, PlanStatus.FAILED, PlanStatus.CANCELLED):
        if not resume:
            logger.error(
                f"Plan is in terminal state {plan.status.value}; "
                "pass --resume to retry remaining phases or create a new plan."
            )
            raise typer.Exit(1)
        if plan.status == PlanStatus.COMPLETED:
            logger.error("Plan is already COMPLETED; nothing to resume.")
            raise typer.Exit(1)
        rolled_back = _reset_plan_for_resume(plan)
        save_plan(plan, data_dir)
        logger.warning(
            f"Resuming plan from terminal state: rolled back {rolled_back} "
            f"phase(s) to PENDING; restarting at phase {plan.current_phase}."
        )
    elif plan.status == PlanStatus.RUNNING:
        if resume:
            # Stuck-RUNNING (previous process crashed): resume reconciles
            # the running phase back to PENDING instead of failing the plan.
            rolled_back = _reset_plan_for_resume(plan)
            save_plan(plan, data_dir)
            logger.warning(
                "Plan was RUNNING on disk with no attached runner; "
                f"resumed and rolled back {rolled_back} phase(s) to PENDING."
            )
        else:
            # A prior process crashed mid-run. Reconcile to FAILED and bail:
            # the user must inspect and pass --resume (or delete) before
            # re-planning.
            plan.status = PlanStatus.FAILED
            plan.error = plan.error or "previous run crashed"
            save_plan(plan, data_dir)
            logger.error(
                "Plan was RUNNING on disk but no runner is attached; marked FAILED. "
                "Pass --resume to retry."
            )
            raise typer.Exit(1)

    try:
        asyncio.run(
            _run_plan(
                settings=settings,
                plan=plan,
                mnemonic=resolved.mnemonic,
                passphrase=resolved.bip39_passphrase or "",
                creation_height=resolved.creation_height,
                data_dir=data_dir,
                network=network,
                backend_type=backend_type,
                rpc_url=rpc_url,
                neutrino_url=neutrino_url,
                directory_servers=directory_servers,
                tor_socks_host=tor_socks_host,
                tor_socks_port=tor_socks_port,
                fee_rate=fee_rate,
                block_target=block_target,
                min_confirmations_between_phases=min_confirmations_between_phases,
                counterparties_override=counterparties,
            )
        )
    except KeyboardInterrupt:
        logger.info("Interrupted by user")
        raise typer.Exit(130)

status_command(wallet_name: Annotated[str | None, typer.Option('--wallet-name', '-w', help='Wallet identifier; defaults to the mnemonic fingerprint')] = None, mnemonic_file: Annotated[Path | None, typer.Option('--mnemonic-file', '-f', help='Path to mnemonic file')] = None, prompt_bip39_passphrase: Annotated[bool, typer.Option('--prompt-bip39-passphrase', help='Prompt for BIP39 passphrase interactively')] = False, data_dir: Annotated[Path | None, typer.Option('--data-dir', envvar='JOINMARKET_DATA_DIR', help='Data directory (default: ~/.joinmarket-ng or $JOINMARKET_DATA_DIR)')] = None, config_file: Annotated[Path | None, typer.Option('--config-file', envvar='JOINMARKET_CONFIG_FILE', help='Config file path (decoupled from data dir). Defaults to <data-dir>/config.toml')] = None, log_level: Annotated[str | None, typer.Option('--log-level', '-l')] = None) -> None

Print the current plan for the given wallet.

Source code in tumbler/src/tumbler/cli.py

@app.command("status")
def status_command(
    wallet_name: Annotated[
        str | None,
        typer.Option(
            "--wallet-name",
            "-w",
            help="Wallet identifier; defaults to the mnemonic fingerprint",
        ),
    ] = None,
    mnemonic_file: Annotated[
        Path | None, typer.Option("--mnemonic-file", "-f", help="Path to mnemonic file")
    ] = None,
    prompt_bip39_passphrase: Annotated[
        bool,
        typer.Option(
            "--prompt-bip39-passphrase",
            help="Prompt for BIP39 passphrase interactively",
        ),
    ] = False,
    data_dir: Annotated[
        Path | None,
        typer.Option(
            "--data-dir",
            envvar="JOINMARKET_DATA_DIR",
            help="Data directory (default: ~/.joinmarket-ng or $JOINMARKET_DATA_DIR)",
        ),
    ] = None,
    config_file: Annotated[
        Path | None,
        typer.Option(
            "--config-file",
            envvar="JOINMARKET_CONFIG_FILE",
            help="Config file path (decoupled from data dir). Defaults to <data-dir>/config.toml",
        ),
    ] = None,
    log_level: Annotated[str | None, typer.Option("--log-level", "-l")] = None,
) -> None:
    """Print the current plan for the given wallet."""
    settings = setup_cli(log_level, data_dir=data_dir, config_file=config_file)
    effective_wallet = _resolve_wallet_name(
        settings, wallet_name, mnemonic_file, prompt_bip39_passphrase
    )
    plan = _load_or_error(effective_wallet, settings.get_data_dir())
    _summarise_plan(plan)