jmwallet.wallet.service

jmwallet.wallet.service

JoinMarket wallet service with mixdepth support.

Attributes

DEFAULT_SCAN_RANGE = 1000 module-attribute

FIDELITY_BOND_BRANCH = 2 module-attribute

__all__ = ['DEFAULT_SCAN_RANGE', 'FIDELITY_BOND_BRANCH', 'WalletService'] module-attribute

Classes

WalletService

Bases: WalletSyncMixin, CoinSelectionMixin, WalletDisplayMixin, WalletSigningMixin

JoinMarket wallet service. Manages BIP84 hierarchical deterministic wallet with mixdepths.

Derivation path: m/84'/0'/{mixdepth}'/{change}/{index} - mixdepth: 0-4 (JoinMarket isolation levels) - change: 0 (external/receive), 1 (internal/change) - index: address index

Source code in jmwallet/src/jmwallet/wallet/service.py

class WalletService(WalletSyncMixin, CoinSelectionMixin, WalletDisplayMixin, WalletSigningMixin):
    """
    JoinMarket wallet service.
    Manages BIP84 hierarchical deterministic wallet with mixdepths.

    Derivation path: m/84'/0'/{mixdepth}'/{change}/{index}
    - mixdepth: 0-4 (JoinMarket isolation levels)
    - change: 0 (external/receive), 1 (internal/change)
    - index: address index
    """

    def __init__(
        self,
        mnemonic: str,
        backend: BlockchainBackend,
        network: str = "mainnet",
        mixdepth_count: int = 5,
        gap_limit: int = 20,
        scan_range: int = DEFAULT_SCAN_RANGE,
        data_dir: Path | None = None,
        passphrase: str = "",
        max_sats_freeze_reuse: int = -1,
        reconstruct_history: bool = True,
    ):
        self.backend = backend
        self.network = network
        self.mixdepth_count = mixdepth_count
        # ``gap_limit`` is the BIP44 trailing-empty threshold (default 20).
        # ``scan_range`` is the descriptor lookahead window imported into
        # Bitcoin Core (default 1000). The two concepts used to be conflated
        # via a ``max(1000, gap_limit * 10)`` formula, dropped in favor of
        # explicit configuration (issue #475).
        self.gap_limit = gap_limit
        self.scan_range = scan_range
        self.data_dir = data_dir
        # Forced address-reuse defense (issue #529): a UTXO that lands on an
        # already-used wallet address is auto-frozen during sync when its value
        # is <= ``max_sats_freeze_reuse`` (or always, when it is -1). 0 disables
        # the behavior. Matches legacy joinmarket-clientserver's
        # ``POLICY.max_sats_freeze_reuse``.
        self.max_sats_freeze_reuse = max_sats_freeze_reuse

        seed = mnemonic_to_seed(mnemonic, passphrase)
        self.master_key = HDKey.from_seed(seed)

        coin_type = 0 if network == "mainnet" else 1
        self.root_path = f"m/84'/{coin_type}'"

        # Log fingerprint for debugging (helps identify passphrase issues)
        fingerprint = self.master_key.derive("m/0").fingerprint.hex()
        # Expose fingerprint as a stable wallet identifier (issue #473).
        # This is the same 8-char hex used by the descriptor wallet name and
        # by the CoinJoin history CSV to scope entries to a specific wallet.
        self.wallet_fingerprint = fingerprint
        logger.info(
            f"Initialized wallet: fingerprint={fingerprint}, "
            f"mixdepths={mixdepth_count}, network={network}, "
            f"passphrase={'(set)' if passphrase else '(none)'}"
        )

        self.address_cache: dict[str, tuple[int, int, int]] = {}
        self._path_cache: dict[tuple[int, int, int], str] = {}
        self.utxo_cache: dict[int, list[UTXOInfo]] = {}
        # Forced-address-reuse defense state (issue #529, hardened for #542).
        #
        # The auto-freeze must distinguish a *genuine* forced reuse (a new coin
        # landing on an address we funded and then emptied) from a perfectly
        # legitimate first-use coin that merely became visible on a later sync
        # (background descriptor rescan still catching up, a transient RPC
        # failure on the first sync, or a descriptor-range upgrade). The
        # persistent ``addresses_with_history`` set is restored at init, so it
        # cannot be used on its own to decide "this address was emptied": a
        # still-funded first-use address is in that set too. Relying on it (via
        # the old one-shot ``_just_initialized`` guard) wrongly froze coins that
        # the first sync had not yet observed (#542).
        #
        # Instead we accumulate the addresses and outpoints we have *positively
        # observed funded*. A coin is treated as forced reuse only when its
        # address was seen funded earlier and is empty again now (see
        # :meth:`_auto_freeze_reused_address_utxos`). These sets are persisted
        # in the metadata store and reseeded below, so the knowledge survives
        # restarts: an address emptied before a restart and refunded after it is
        # still frozen, while coins that predate the restart (in the persisted
        # seen-outpoint set) and late-discovered first-use coins (never
        # persisted as observed-funded) are left spendable.
        self._observed_funded_addresses: set[str] = set()
        self._observed_outpoints: set[str] = set()
        # Guards the once-per-process import-label reconstruction pass (see
        # WalletSyncMixin.reconstruct_imported_labels). Coins received while
        # running are either this wallet's own CoinJoins (recorded in history)
        # or genuine deposits, so only the imported backlog needs scanning.
        self._imported_labels_scanned: bool = False
        # Guards the once-per-process import-history reconstruction pass (see
        # WalletSyncMixin.reconstruct_imported_history). The automatic pass
        # only fires for wallets with no recorded history (seed imports);
        # ``reconstruct_history`` maps the [wallet] reconstruct_history config
        # toggle.
        self._imported_history_scanned: bool = False
        # True once an empty-history wallet has entered the imported-history
        # workflow, even if reconstruction was deferred for a Core rescan.
        # This keeps a protocol row written during that rescan from cancelling
        # the pending backfill later in the same process.
        self._imported_history_started: bool = False
        self.reconstruct_history_enabled: bool = reconstruct_history

        # UTXO + address metadata store (BIP-329 JSONL). Frozen UTXO state,
        # output labels, and the persistent "addresses with on-chain history"
        # set all live in the same per-wallet ``wallet_metadata_<fp>.jsonl``
        # file. Partitioning by fingerprint is mandatory: pre-0.30.0 builds
        # used a shared ``wallet_metadata.jsonl`` per data_dir, which leaked
        # one wallet's used-address set and frozen-UTXO state into any
        # other wallet opened in the same directory.
        self.metadata_store: UTXOMetadataStore | None = None
        if data_dir is not None:
            # Only pre-derive owned addresses when the one-shot migration
            # from the legacy shared file is actually going to run. After
            # the first open, the per-wallet file exists and the
            # migration is skipped, so the typical hot path pays nothing.
            from jmcore.paths import get_wallet_metadata_path

            per_wallet_path = get_wallet_metadata_path(data_dir, fingerprint=fingerprint)
            shared_path = get_wallet_metadata_path(data_dir, fingerprint=None)
            owned_addresses: set[str] | None = None
            if not per_wallet_path.exists() and shared_path.exists():
                # Derivation is pure compute (no backend RPCs); for the
                # default scan_range=1000 / mixdepth_count=5 this is
                # roughly 10k BIP32 derivations and completes in well
                # under a second. The derived addresses are also seeded
                # into ``self.address_cache`` so subsequent lookups
                # skip re-derivation.
                owned_addresses = set()
                for mixdepth in range(self.mixdepth_count):
                    for change in (0, 1):
                        for index in range(self.scan_range):
                            owned_addresses.add(self.get_address(mixdepth, change, index))
            self.metadata_store = load_metadata_store(
                data_dir,
                fingerprint=fingerprint,
                owned_addresses=owned_addresses,
            )

        # Track addresses that have ever had UTXOs (including spent ones).
        # Used to label addresses as "used-empty" vs "new" and, critically,
        # to prevent reissuing a previously-funded deposit address. Backed by
        # the metadata store so the knowledge survives across runs (light
        # clients like Neutrino and Bitcoin Core's address-book-bound
        # ``listreceivedbyaddress`` cannot always rediscover spent-then-empty
        # addresses from scratch).
        self.addresses_with_history: set[str] = set()
        if self.metadata_store is not None:
            self.addresses_with_history.update(self.metadata_store.get_used_addresses())
            self._migrate_legacy_address_history(data_dir)
            # Reseed the forced-address-reuse observation sets so the defense
            # survives restarts (issue #559).
            self._observed_funded_addresses.update(
                self.metadata_store.get_observed_funded_addresses()
            )
            self._observed_outpoints.update(self.metadata_store.get_seen_outpoints())

        # Track addresses currently reserved for in-progress CoinJoin sessions
        # These addresses have been shared with a taker but the CoinJoin hasn't
        # completed yet. They must not be reused until the session ends.
        self.reserved_addresses: set[str] = set()
        # Track receive addresses that were already handed out via API/CLI.
        # Even if they do not appear on-chain yet, we should not reissue them.
        self.issued_receive_addresses: set[str] = set()
        # Reserved (set-aside) deposit addresses -> optional user label.
        # Persisted in the metadata store (``jm:reserved`` records) so handed-out
        # and user-labeled deposit addresses survive restarts, are never
        # reissued, and can be shown with their label in the extended view.
        self.reserved_address_labels: dict[str, str] = {}
        if self.metadata_store is not None:
            self.reserved_address_labels.update(self.metadata_store.get_reserved_labels())
            # Feed the in-memory picker skip-set so reserved addresses are not
            # reissued as the next unused deposit address after a restart.
            self.issued_receive_addresses.update(self.reserved_address_labels.keys())
        # Cache for fidelity bond locktimes (address -> locktime)
        self.fidelity_bond_locktime_cache: dict[str, int] = {}
        # Lazily-built cache of every canonical fidelity-bond address (all
        # 960 timenumbers) mapped to its (locktime, timenumber). Populated on
        # first use by ``WalletSyncMixin._canonical_bond_address_map``; see
        # that method for why this exists (recognizing bond UTXOs Bitcoin
        # Core already tracks even when the local registry has no matching
        # entry, issue: fidelity bonds invisible after per-wallet registry
        # partition / #492 migration gaps).
        self._canonical_bond_addresses: dict[str, tuple[int, int]] | None = None

        # One-shot migration of the legacy shared ``fidelity_bonds.json``
        # registry into a per-wallet ``fidelity_bonds_<fp>.json`` file
        # (issue #492). Same shape as the metadata-store migration above:
        # only runs when the per-wallet file is missing AND the legacy
        # file exists, so the typical hot path is a single ``Path.exists``
        # check.
        if data_dir is not None:
            self._migrate_legacy_bond_registry(data_dir)

        # Resolve reserved deposit addresses to their derivation path and seed
        # the address cache. The deposit-address pickers key on the cache to
        # map an address to its index, so this lets them advance past a
        # reserved address even before a full sync has populated the cache
        # (after a restart the reservation is loaded from disk, but the
        # address itself has not been derived yet). Reserved addresses are few
        # and typically at low indices, so this is cheap.
        for reserved_addr in list(self.reserved_address_labels):
            if reserved_addr not in self.address_cache:
                try:
                    self._find_address_path(reserved_addr, max_scan=self.scan_range)
                except Exception as exc:  # pragma: no cover - defensive
                    logger.debug(f"Could not resolve reserved address {reserved_addr}: {exc}")

    def _migrate_legacy_address_history(self, data_dir: Path | None) -> None:
        """Fold a legacy ``address_history_<fingerprint>.jsonl`` file into the
        unified metadata store, then remove it.

        Pre-0.30.0 builds shipped a brief intermediate format that stored the
        privacy-critical "used addresses" set in its own JSONL file. The
        current architecture keeps it inside ``wallet_metadata.jsonl`` as
        BIP-329 ``addr`` records. This one-shot migration runs at startup;
        once the legacy file is consumed it is unlinked so subsequent runs
        skip the check cheaply.
        """
        if data_dir is None or self.metadata_store is None:
            return
        safe_fp = self.wallet_fingerprint.strip().lower()
        if not safe_fp.isalnum():
            return
        legacy_path = data_dir / f"address_history_{safe_fp}.jsonl"
        if not legacy_path.exists():
            return
        try:
            text = legacy_path.read_text(encoding="utf-8")
        except OSError as exc:
            logger.warning(f"Failed to read legacy address history {legacy_path}: {exc}")
            return
        migrated: list[str] = []
        for raw in text.splitlines():
            line = raw.strip()
            if not line or line.startswith("#"):
                continue
            try:
                value = json.loads(line)
            except json.JSONDecodeError:
                continue
            if isinstance(value, str) and value:
                migrated.append(value)
        if migrated:
            self.metadata_store.mark_addresses_used(migrated, origin="legacy")
            self.addresses_with_history.update(migrated)
            logger.info(
                f"Migrated {len(migrated)} address(es) from legacy "
                f"{legacy_path.name} into wallet_metadata.jsonl"
            )
        try:
            legacy_path.unlink()
        except OSError as exc:  # pragma: no cover - defensive
            logger.warning(f"Could not remove legacy {legacy_path.name}: {exc}")

    def _migrate_legacy_bond_registry(self, data_dir: Path) -> None:
        """Claim entries from the legacy ``fidelity_bonds.json`` into a
        per-wallet ``fidelity_bonds_<fp>.json`` file (issue #492).

        Pre-0.30.0 builds wrote every wallet's fidelity bonds into a
        single shared ``fidelity_bonds.json`` under the data directory.
        A side effect was that ``jm-wallet list-bonds`` (and the maker
        bot) saw bonds belonging to other wallets opened from the same
        directory. To restore per-wallet isolation we partition the file
        per wallet fingerprint, matching the
        ``wallet_metadata_<fp>.jsonl`` precedent.

        Migration is wallet-aware: for each entry in the legacy file the
        bond's stored pubkey is compared against the pubkey re-derived
        from the open wallet at ``bond.path`` (or, when the path is the
        canonical fidelity-bond branch, from the timenumber derived from
        ``bond.locktime``). Matching entries are claimed by this wallet
        and written to the per-wallet file. Non-matching entries are
        left in the legacy file so other wallets can claim them on their
        next open. The legacy file is removed once empty.
        """
        from jmwallet.wallet.bond_registry import (
            make_wallet_ownership_predicate,
            migrate_legacy_registry,
        )

        predicate = make_wallet_ownership_predicate(self.master_key, self.root_path)
        migrate_legacy_registry(data_dir, self.wallet_fingerprint, predicate)

    # -- Key derivation & address generation (Group A) ----------------------

    def get_address(self, mixdepth: int, change: int, index: int) -> str:
        """Get address for given path"""
        if mixdepth >= self.mixdepth_count:
            raise ValueError(f"Mixdepth {mixdepth} exceeds maximum {self.mixdepth_count}")

        path_key = (mixdepth, change, index)
        cached = self._path_cache.get(path_key)
        if cached is not None:
            self.address_cache[cached] = path_key
            return cached

        path = f"{self.root_path}/{mixdepth}'/{change}/{index}"
        key = self.master_key.derive(path)
        address = key.get_address(self.network)

        self.address_cache[address] = (mixdepth, change, index)
        self._path_cache[path_key] = address

        return address

    def get_receive_address(self, mixdepth: int, index: int) -> str:
        """Get external (receive) address"""
        return self.get_address(mixdepth, 0, index)

    def get_change_address(self, mixdepth: int, index: int) -> str:
        """Get internal (change) address"""
        return self.get_address(mixdepth, 1, index)

    def get_account_xpub(self, mixdepth: int) -> str:
        """
        Get the extended public key (xpub) for a mixdepth account.

        Derives the key at path m/84'/coin'/mixdepth' and returns its xpub.
        This xpub can be used in Bitcoin Core descriptors for efficient scanning.

        Args:
            mixdepth: The mixdepth (account) number (0-4)

        Returns:
            xpub/tpub string for the account
        """
        account_path = f"{self.root_path}/{mixdepth}'"
        account_key = self.master_key.derive(account_path)
        return account_key.get_xpub(self.network)

    def get_account_zpub(self, mixdepth: int) -> str:
        """
        Get the BIP84 extended public key (zpub) for a mixdepth account.

        Derives the key at path m/84'/coin'/mixdepth' and returns its zpub.
        zpub explicitly indicates this is a native segwit (P2WPKH) wallet.

        Args:
            mixdepth: The mixdepth (account) number (0-4)

        Returns:
            zpub/vpub string for the account
        """
        account_path = f"{self.root_path}/{mixdepth}'"
        account_key = self.master_key.derive(account_path)
        return account_key.get_zpub(self.network)

    def get_scan_descriptors(self, scan_range: int = DEFAULT_SCAN_RANGE) -> list[dict[str, Any]]:
        """
        Generate descriptors for efficient UTXO scanning with Bitcoin Core.

        Creates wpkh() descriptors with xpub and range for all mixdepths,
        both external (receive) and internal (change) addresses.

        Using descriptors with ranges is much more efficient than scanning
        individual addresses, as Bitcoin Core can scan the entire range in
        a single pass through the UTXO set.

        Args:
            scan_range: Maximum index to scan (default 1000, Bitcoin Core's default)

        Returns:
            List of descriptor dicts for use with scantxoutset:
            [{"desc": "wpkh(xpub.../0/*)", "range": [0, 999]}, ...]
        """
        descriptors = []

        for mixdepth in range(self.mixdepth_count):
            xpub = self.get_account_xpub(mixdepth)

            # External (receive) addresses: .../0/*
            descriptors.append({"desc": f"wpkh({xpub}/0/*)", "range": [0, scan_range - 1]})

            # Internal (change) addresses: .../1/*
            descriptors.append({"desc": f"wpkh({xpub}/1/*)", "range": [0, scan_range - 1]})

        logger.debug(
            f"Generated {len(descriptors)} descriptors for {self.mixdepth_count} mixdepths "
            f"with range [0, {scan_range - 1}]"
        )
        return descriptors

    def get_fidelity_bond_key(self, index: int, locktime: int) -> HDKey:
        """
        Get the HD key for a fidelity bond.

        Fidelity bond path: m/84'/coin'/0'/2/timenumber

        In the JoinMarket protocol, the BIP32 child index for fidelity bonds
        is the **timenumber** (0-959), NOT a separate address index. Each
        timenumber maps 1:1 to a locktime (1st of month, Jan 2020 - Dec 2099).

        For backward compatibility, the ``index`` parameter is still accepted
        but is **ignored** when ``locktime`` is a valid timenumber locktime.
        The timenumber is computed from the locktime and used as the child index.

        Args:
            index: Legacy address index (ignored when locktime is valid).
                   Kept for API compatibility.
            locktime: Unix timestamp for the timelock. Must be a valid
                      timenumber locktime (1st of month, midnight UTC).

        Returns:
            HDKey for the fidelity bond
        """
        from jmcore.timenumber import timestamp_to_timenumber

        # The BIP32 child index is the timenumber derived from the locktime,
        # matching the reference JoinMarket implementation.
        timenumber = timestamp_to_timenumber(locktime)
        path = f"{self.root_path}/0'/{FIDELITY_BOND_BRANCH}/{timenumber}"
        return self.master_key.derive(path)

    def get_fidelity_bond_address(self, index: int, locktime: int) -> str:
        """
        Get a fidelity bond P2WSH address.

        Creates a timelocked script: <locktime> OP_CLTV OP_DROP <pubkey> OP_CHECKSIG
        wrapped in P2WSH.

        The ``index`` parameter is a legacy argument and is **ignored**; the
        BIP32 child index is always the timenumber derived from ``locktime``.

        Args:
            index: Legacy address index (ignored; timenumber is used instead)
            locktime: Unix timestamp for the timelock

        Returns:
            P2WSH address for the fidelity bond
        """
        from jmcore.timenumber import timestamp_to_timenumber

        key = self.get_fidelity_bond_key(index, locktime)
        pubkey_hex = key.get_public_key_bytes(compressed=True).hex()

        # Create the timelock script
        script = mk_freeze_script(pubkey_hex, locktime)

        # Convert to P2WSH address
        address = script_to_p2wsh_address(script, self.network)

        # Cache with timenumber as the index (matches BIP32 child index)
        timenumber = timestamp_to_timenumber(locktime)
        self.address_cache[address] = (0, FIDELITY_BOND_BRANCH, timenumber)
        # Also store the locktime in a separate cache for fidelity bonds
        self.fidelity_bond_locktime_cache[address] = locktime

        logger.trace(f"Created fidelity bond address {address} with locktime {locktime}")
        return address

    def get_fidelity_bond_script(self, index: int, locktime: int) -> bytes:
        """
        Get the redeem script for a fidelity bond.

        The ``index`` parameter is a legacy argument and is **ignored**; the
        BIP32 child index is always the timenumber derived from ``locktime``.

        Args:
            index: Legacy address index (ignored; timenumber is used instead)
            locktime: Unix timestamp for the timelock

        Returns:
            Timelock redeem script bytes
        """
        key = self.get_fidelity_bond_key(index, locktime)
        pubkey_hex = key.get_public_key_bytes(compressed=True).hex()
        return mk_freeze_script(pubkey_hex, locktime)

    def get_locktime_for_address(self, address: str) -> int | None:
        """
        Get the locktime for a fidelity bond address.

        Args:
            address: The fidelity bond address

        Returns:
            Locktime as Unix timestamp, or None if not a fidelity bond address
        """
        return self.fidelity_bond_locktime_cache.get(address)

    def get_private_key(self, mixdepth: int, change: int, index: int) -> bytes:
        """Get private key for given path"""
        path = f"{self.root_path}/{mixdepth}'/{change}/{index}"
        key = self.master_key.derive(path)
        return key.get_private_key_bytes()

    def get_key_for_address(self, address: str) -> HDKey | None:
        """Get HD key for a known address"""
        path_info = self.address_cache.get(address)
        if path_info is None:
            path_info = self.address_cache.get(address.lower())
        if path_info is None:
            return None

        mixdepth, change, index = path_info
        path = f"{self.root_path}/{mixdepth}'/{change}/{index}"
        return self.master_key.derive(path)

    # -- Balance & UTXO queries (Group G) -----------------------------------

    async def get_balance(
        self, mixdepth: int, include_fidelity_bonds: bool = True, min_confirmations: int = 0
    ) -> int:
        """Get balance for a mixdepth.

        Args:
            mixdepth: Mixdepth to get balance for
            include_fidelity_bonds: If True (default), include fidelity bond UTXOs.
                                    If False, exclude fidelity bond UTXOs.
            min_confirmations: Minimum confirmations required (default: 0).

        Note:
            Frozen UTXOs are excluded from balance calculations.
        """
        if mixdepth not in self.utxo_cache:
            await self.sync_mixdepth(mixdepth)

        utxos = self.utxo_cache.get(mixdepth, [])
        utxos = [u for u in utxos if not u.frozen]
        if not include_fidelity_bonds:
            utxos = [u for u in utxos if not u.is_fidelity_bond]
        if min_confirmations > 0:
            utxos = [u for u in utxos if u.confirmations >= min_confirmations]
        return sum(utxo.value for utxo in utxos)

    async def get_balance_for_offers(
        self, mixdepth: int, min_confirmations: int = 0, *, restrict_md0: bool = True
    ) -> int:
        """Get balance available for maker offers (excludes fidelity bond UTXOs).

        Fidelity bonds should never be automatically spent in CoinJoins,
        so makers must exclude them when calculating available offer amounts.

        For mixdepth 0 (when ``restrict_md0`` is True), UTXOs that are **not**
        CoinJoin outputs are restricted to a single UTXO to avoid linking
        deposits or fidelity bonds.  CoinJoin outputs (``label == "cj-out"``)
        are exempt because they already have CoinJoin privacy and can be
        safely merged.

        The effective balance is therefore::

            max(sum_of_cj_outputs, largest_non_cj_output)

        When ``restrict_md0`` is False (opt-in via config), mixdepth 0 is
        treated the same as any other mixdepth.
        """
        if mixdepth == 0 and restrict_md0:
            if mixdepth not in self.utxo_cache:
                await self.sync_mixdepth(mixdepth)
            utxos = self.utxo_cache.get(mixdepth, [])
            eligible = [
                u
                for u in utxos
                if not u.frozen and not u.is_fidelity_bond and u.confirmations >= min_confirmations
            ]
            if not eligible:
                return 0

            cj_pool = sum(u.value for u in eligible if u.label == "cj-out")
            non_cj = [u for u in eligible if u.label != "cj-out"]
            largest_single = max((u.value for u in non_cj), default=0)
            return max(cj_pool, largest_single)

        return await self.get_balance(
            mixdepth, include_fidelity_bonds=False, min_confirmations=min_confirmations
        )

    async def get_utxos(self, mixdepth: int) -> list[UTXOInfo]:
        """Get UTXOs for a mixdepth, syncing if not cached."""
        if mixdepth not in self.utxo_cache:
            await self.sync_mixdepth(mixdepth)
        return self.utxo_cache.get(mixdepth, [])

    def find_utxo_by_address(self, address: str) -> UTXOInfo | None:
        """
        Find a UTXO by its address across all mixdepths.

        This is useful for matching CoinJoin outputs to history entries.
        Returns the first matching UTXO found, or None if address not found.

        Args:
            address: Bitcoin address to search for

        Returns:
            UTXOInfo if found, None otherwise
        """
        for mixdepth in range(self.mixdepth_count):
            utxos = self.utxo_cache.get(mixdepth, [])
            for utxo in utxos:
                if utxo.address == address:
                    return utxo
        return None

    async def get_total_balance(
        self, include_fidelity_bonds: bool = True, min_confirmations: int = 0
    ) -> int:
        """Get the spendable balance across all mixdepths.

        Despite the name, this is the *spendable* total: frozen UTXOs are
        always excluded, and fidelity bonds are excluded when
        ``include_fidelity_bonds`` is False. Callers that need the grand total
        (including frozen funds) must add the frozen amount back themselves.

        Args:
            include_fidelity_bonds: If True (default), include fidelity bond UTXOs.
                                    If False, exclude fidelity bond UTXOs.
            min_confirmations: Minimum confirmations required (default: 0).

        Note:
            Frozen UTXOs are excluded from balance calculations.
        """
        total = 0
        for mixdepth in range(self.mixdepth_count):
            balance = await self.get_balance(
                mixdepth,
                include_fidelity_bonds=include_fidelity_bonds,
                min_confirmations=min_confirmations,
            )
            total += balance
        return total

    async def get_fidelity_bond_balance(self, mixdepth: int) -> int:
        """Get balance of fidelity bond UTXOs for a mixdepth.

        Note:
            Unlike spendable-balance helpers, the ``frozen`` flag is **not**
            applied here. A fidelity bond is already excluded from automatic
            coin selection by virtue of being a timelocked bond, so its
            ``frozen`` flag is orthogonal to its informational value. The
            maker advertises the bond, ``list-bonds`` reports it as ACTIVE,
            and the extended wallet view counts it regardless of ``frozen``;
            this helper backs the basic ``jm-wallet info`` ``(+... FB)``
            annotation and must report the same bond value so the views stay
            consistent (see issue: bond hidden from basic info after freeze).
        """
        if mixdepth not in self.utxo_cache:
            await self.sync_mixdepth(mixdepth)

        utxos = self.utxo_cache.get(mixdepth, [])
        return sum(utxo.value for utxo in utxos if utxo.is_fidelity_bond)

    # -- Address index management (Group I) ---------------------------------

    def get_next_address_index(self, mixdepth: int, change: int) -> int:
        """
        Get next unused address index for mixdepth/change.

        Returns the highest index + 1 among all addresses that have ever been used,
        ensuring we never reuse addresses. An address is considered "used" if it:
        - Has current UTXOs
        - Had UTXOs in the past (tracked in addresses_with_history)
        - Appears in CoinJoin history (even if never funded)

        We always return one past the highest used index, even if lower indices
        appear unused. Those may have been skipped for a reason (e.g., shared in
        a failed CoinJoin, or spent in an internal transfer).
        """
        max_index = -1

        # Check addresses with current UTXOs
        utxos = self.utxo_cache.get(mixdepth, [])
        for utxo in utxos:
            if utxo.address in self.address_cache:
                md, ch, idx = self.address_cache[utxo.address]
                if md == mixdepth and ch == change and idx > max_index:
                    max_index = idx

        # Check addresses that ever had blockchain activity (including spent)
        for address in self.addresses_with_history:
            if address in self.address_cache:
                md, ch, idx = self.address_cache[address]
                if md == mixdepth and ch == change and idx > max_index:
                    max_index = idx

        # Check CoinJoin history for addresses that may have been shared
        # but never received funds (e.g., failed CoinJoins)
        if self.data_dir:
            from jmwallet.history import get_used_addresses

            cj_addresses = get_used_addresses(
                self.data_dir, wallet_fingerprint=self.wallet_fingerprint
            )
            self._prune_reserved_addresses(cj_addresses | self.addresses_with_history)
            for address in cj_addresses:
                if address in self.address_cache:
                    md, ch, idx = self.address_cache[address]
                    if md == mixdepth and ch == change and idx > max_index:
                        max_index = idx

        # Check addresses reserved for in-progress CoinJoin sessions
        # These have been shared with takers but the session hasn't completed yet
        for address in self.reserved_addresses:
            if address in self.address_cache:
                md, ch, idx = self.address_cache[address]
                if md == mixdepth and ch == change and idx > max_index:
                    max_index = idx

        # Check receive addresses that were already issued to callers.
        # This prevents repeated GET /address/new/{mixdepth} calls from
        # returning the same address when no on-chain history exists yet.
        for address in self.issued_receive_addresses:
            if address in self.address_cache:
                md, ch, idx = self.address_cache[address]
                if md == mixdepth and ch == change and idx > max_index:
                    max_index = idx

        return max_index + 1

    def _prune_reserved_addresses(self, persisted_addresses: set[str]) -> None:
        """Drop reserved addresses that are already tracked by durable history.

        ``reserved_addresses`` only needs to keep addresses that were handed out in
        this runtime but are not yet persisted in history. Once an address appears
        in CoinJoin/chain history, keeping it in-memory is redundant.
        """
        if not self.reserved_addresses:
            return

        before = len(self.reserved_addresses)
        self.reserved_addresses.difference_update(persisted_addresses)
        removed = before - len(self.reserved_addresses)
        if removed > 0:
            logger.debug(f"Pruned {removed} reserved addresses now covered by persisted history")

    def reserve_addresses(self, addresses: set[str]) -> None:
        """
        Reserve addresses for an in-progress CoinJoin session.

        Once addresses are shared with a taker (in !ioauth message), they must not
        be reused even if the CoinJoin fails. This method marks addresses as reserved
        so get_next_address_index() will skip past them.

        Note: Addresses stay reserved until the wallet is restarted, since they may
        have been logged by counterparties. The CoinJoin history file provides
        persistent tracking across restarts.

        Args:
            addresses: Set of addresses to reserve (typically cj_address + change_address)
        """
        self.reserved_addresses.update(addresses)
        logger.debug(f"Reserved {len(addresses)} addresses: {addresses}")

    async def sync(self) -> dict[int, list[UTXOInfo]]:
        """Sync wallet (alias for sync_all for backward compatibility)."""
        return await self.sync_all()

    def get_new_address(self, mixdepth: int) -> str:
        """Get next unused receive address for a mixdepth.

        Synchronous fast path: returns the address at
        ``get_next_address_index(mixdepth, 0)``. This relies on the
        sync-layer ``addresses_with_history`` being complete; if the
        last bulk enumeration was truncated by an RPC failure, this
        method may return a previously-funded address.

        Privacy-critical callers (CLI ``info`` / daemon address
        endpoints) should prefer :meth:`get_new_address_verified`,
        which adds a per-candidate ``getreceivedbyaddress`` check.
        """
        next_index = self.get_next_address_index(mixdepth, 0)
        address = self.get_receive_address(mixdepth, next_index)
        self.reserve_address(address)
        return address

    async def get_new_address_verified(self, mixdepth: int) -> str:
        """Async deposit-address picker with on-chain verification.

        Wraps :meth:`get_next_safe_deposit_address` and reserves the chosen
        address (persisted when a ``data_dir`` is configured) so it is never
        reissued, even across restarts. Use this from any async code path that
        exposes a deposit address to users or peers (``jm-wallet info``,
        jmwalletd ``/wallet/address/new``, maker/taker deposit prompts).
        """
        address, _ = await self.get_next_safe_deposit_address(mixdepth)
        self.reserve_address(address)
        return address

    def reserve_address(self, address: str, label: str = "") -> None:
        """Reserve (set aside) a deposit address so it is never reissued.

        Records the address in the in-memory skip-set consulted by the
        deposit-address pickers and, when a ``data_dir`` is configured,
        persists a ``jm:reserved`` record (with the optional user label) to the
        metadata store so the reservation survives restarts. Reserved addresses
        are hidden from the concise ``jm-wallet info`` view and shown with their
        label in the extended view.
        """
        if not address:
            return
        self.issued_receive_addresses.add(address)
        self.reserved_address_labels[address] = label or ""
        store = getattr(self, "metadata_store", None)
        if store is not None:
            try:
                store.reserve_address(address, label or "")
            except Exception as exc:  # pragma: no cover - disk failures are rare
                logger.warning(f"Failed to persist reserved address {address}: {exc}")

    def unreserve_address(self, address: str) -> bool:
        """Remove a reservation so the address may be reissued.

        Note: an address that has real on-chain history is still never
        reissued (the pickers also consult ``addresses_with_history``); this
        only clears the "set aside" marker and its label.
        """
        changed = self.reserved_address_labels.pop(address, None) is not None
        self.issued_receive_addresses.discard(address)
        store = getattr(self, "metadata_store", None)
        if store is not None:
            try:
                if store.unreserve_address(address):
                    changed = True
            except Exception as exc:  # pragma: no cover - disk failures are rare
                logger.warning(f"Failed to remove reserved address {address}: {exc}")
        return changed

    def is_address_reserved(self, address: str) -> bool:
        """Return True if ``address`` has been reserved/set aside by the user."""
        return address in self.reserved_address_labels

    def get_reserved_addresses(self) -> dict[str, str]:
        """Return a copy of the reserved address -> user label mapping."""
        return dict(self.reserved_address_labels)

    async def close(self) -> None:
        """Close backend connection"""
        await self.backend.close()

    # -- UTXO metadata (Group J) -------------------------------------------

    def _auto_freeze_reused_address_utxos(
        self,
        observed_funded_addresses: set[str],
        observed_outpoints: set[str],
        prior_funded_addresses: set[str],
    ) -> int:
        """Auto-freeze UTXOs that landed on an already-spent (empty) used address.

        Defends against forced address-reuse (dust) attacks: an adversary pays a
        small amount to an address the wallet has already used and emptied,
        hoping the new coin gets co-spent and links the wallet's coins via the
        common-input-ownership heuristic. Per
        https://en.bitcoin.it/wiki/Privacy#Forced_address_reuse, coins that land
        on an already-used *empty* address should never be spent (we freeze
        them); coins on an address that still holds funds should instead be
        fully spent together, so those are left untouched.

        A UTXO is auto-frozen only when ALL of the following hold:

        * Its outpoint is NOT in ``observed_outpoints`` -- it is a coin this
          process is seeing for the first time, never one we have already
          accounted for (this also makes the check robust to a transient sync
          that momentarily lost and then re-found the same coin).
        * Its address IS in ``observed_funded_addresses`` -- we positively
          observed this address holding a coin on an earlier sync. This is the
          crucial guard against #542: we never freeze a first-use coin that was
          merely *discovered late* (e.g. by a background rescan), because we
          would not have seen its address funded before.
        * Its address is NOT in ``prior_funded_addresses`` -- the address held
          no UTXO at the start of this sync, i.e. it was emptied before this
          arrival. If the address still holds funds the privacy-correct action
          is to fully spend them together, so those are left untouched. This is
          the key difference from legacy joinmarket-clientserver, which froze
          reuse on any used address.
        * It passes the value filter: ``max_sats_freeze_reuse == -1`` freezes
          all such reuse, a positive ``N`` freezes only ``value <= N`` sats, and
          ``0`` disables the behavior entirely.

        A UTXO that already has a metadata record (e.g. one the user
        deliberately unfroze, which keeps a labeled record) is left untouched,
        so an explicit unfreeze is never overridden. Fidelity bonds (timelocked,
        on the dedicated bond branch) are skipped.

        Returns the number of UTXOs newly frozen.
        """
        if self.metadata_store is None:
            return 0
        threshold = self.max_sats_freeze_reuse
        if threshold == 0:
            return 0
        if not observed_funded_addresses:
            return 0

        frozen_now = 0
        for utxos in self.utxo_cache.values():
            for utxo in utxos:
                if utxo.is_fidelity_bond:
                    continue
                outpoint = utxo.outpoint
                # Coins we have already observed (the original deposit, coins
                # present at startup, or a coin transiently lost then re-found)
                # are never auto-frozen -- only genuinely new arrivals are.
                if outpoint in observed_outpoints:
                    continue
                # Only freeze a new coin on an address we have *positively seen
                # funded before* and that is empty again now. A first-use coin
                # surfaced by a later sync (background rescan, transient RPC
                # failure, descriptor-range upgrade) was never observed funded,
                # so it is left spendable (issue #542).
                if utxo.address not in observed_funded_addresses:
                    continue
                if utxo.address in prior_funded_addresses:
                    continue
                if threshold != -1 and utxo.value > threshold:
                    continue
                # Skip UTXOs the wallet already tracks (already frozen, labeled,
                # locked, or previously evaluated): never override a user's
                # explicit unfreeze of a reuse UTXO.
                if self.metadata_store.has_record(outpoint):
                    continue
                self.metadata_store.freeze(outpoint, label=AUTO_FREEZE_REUSE_LABEL)
                utxo.frozen = True
                frozen_now += 1
                logger.warning(
                    "Auto-froze UTXO to prevent forced address reuse: "
                    f"{outpoint} ({utxo.value} sats at {utxo.address[:16]}...). "
                    "Unfreeze with 'jm-wallet unfreeze' if intentional."
                )

        if frozen_now:
            logger.warning(
                f"Auto-froze {frozen_now} UTXO(s) on reused empty addresses "
                "(forced-address-reuse defense)."
            )
        return frozen_now

    def _apply_frozen_state(self) -> None:
        """Apply frozen state from metadata store to all cached UTXOs.

        Called after sync operations to mark UTXOs that are frozen according
        to the persisted metadata. Also applies labels from metadata.

        Re-reads the metadata file from disk on each call to pick up changes
        made by other processes (e.g., ``jm-wallet freeze`` while maker is running).
        """
        if self.metadata_store is None:
            return

        # Re-read from disk to pick up changes from other processes
        self.metadata_store.load()

        frozen_outpoints = self.metadata_store.get_frozen_outpoints()

        frozen_count = 0
        for utxos in self.utxo_cache.values():
            for utxo in utxos:
                outpoint = utxo.outpoint
                utxo.frozen = outpoint in frozen_outpoints
                if utxo.frozen:
                    frozen_count += 1
                # Apply label from metadata if not already set
                stored_label = self.metadata_store.get_label(outpoint)
                if stored_label is not None and utxo.label is None:
                    utxo.label = stored_label

        if frozen_count > 0:
            logger.debug(f"Applied frozen state to {frozen_count} UTXO(s)")

    def freeze_utxo(self, outpoint: str) -> None:
        """Freeze a UTXO by outpoint (persisted to disk).

        Args:
            outpoint: Outpoint string in ``txid:vout`` format.

        Raises:
            RuntimeError: If no metadata store is available (no data_dir).
        """
        if self.metadata_store is None:
            raise RuntimeError("Cannot freeze UTXOs without a data directory")
        self.metadata_store.freeze(outpoint)
        # Update the in-memory UTXO cache
        for utxos in self.utxo_cache.values():
            for utxo in utxos:
                if utxo.outpoint == outpoint:
                    utxo.frozen = True
                    return

    def unfreeze_utxo(self, outpoint: str) -> None:
        """Unfreeze a UTXO by outpoint (persisted to disk).

        Args:
            outpoint: Outpoint string in ``txid:vout`` format.

        Raises:
            RuntimeError: If no metadata store is available (no data_dir).
        """
        if self.metadata_store is None:
            raise RuntimeError("Cannot unfreeze UTXOs without a data directory")
        self.metadata_store.unfreeze(outpoint)
        # Update the in-memory UTXO cache
        for utxos in self.utxo_cache.values():
            for utxo in utxos:
                if utxo.outpoint == outpoint:
                    utxo.frozen = False
                    return

    # -- Temporary CoinJoin input locks (cross-process) ----------------------

    def get_locked_input_outpoints(self) -> set[tuple[str, int]]:
        """Return ``(txid, vout)`` inputs currently locked by any in-flight round.

        Re-reads the on-disk metadata so locks written by other processes
        (another taker round, or a maker serving a different taker) are visible
        right before coin selection. Returns an empty set when no metadata store
        is configured (no data directory).
        """
        if self.metadata_store is None:
            return set()
        self.metadata_store.load()
        locked: set[tuple[str, int]] = set()
        for ref in self.metadata_store.get_locked_outpoints():
            txid, _, vout = ref.rpartition(":")
            if txid and vout.isdigit():
                locked.add((txid, int(vout)))
        return locked

    def reserve_coinjoin_inputs(
        self,
        outpoints: set[tuple[str, int]],
        ttl: float = DEFAULT_COINJOIN_LOCK_TTL,
    ) -> bool:
        """Atomically lock ``outpoints`` for an in-flight CoinJoin.

        Returns True if all were locked, False on conflict (another round
        already holds one of them). When no metadata store is configured the
        call is a no-op that returns True (locking is best-effort persistence).
        """
        if self.metadata_store is None or not outpoints:
            return True
        refs = [f"{txid}:{vout}" for txid, vout in outpoints]
        return self.metadata_store.try_lock_outpoints(refs, ttl=ttl)

    def release_coinjoin_inputs(self, outpoints: set[tuple[str, int]]) -> None:
        """Release CoinJoin locks held on ``outpoints`` (no-op if none)."""
        if self.metadata_store is None or not outpoints:
            return
        refs = [f"{txid}:{vout}" for txid, vout in outpoints]
        self.metadata_store.release_outpoints(refs)

    def toggle_freeze_utxo(self, outpoint: str) -> bool:
        """Toggle frozen state of a UTXO by outpoint (persisted to disk).

        Args:
            outpoint: Outpoint string in ``txid:vout`` format.

        Returns:
            True if now frozen, False if now unfrozen.

        Raises:
            RuntimeError: If no metadata store is available (no data_dir).
        """
        if self.metadata_store is None:
            raise RuntimeError("Cannot toggle freeze without a data directory")
        now_frozen = self.metadata_store.toggle_freeze(outpoint)
        # Update the in-memory UTXO cache
        for utxos in self.utxo_cache.values():
            for utxo in utxos:
                if utxo.outpoint == outpoint:
                    utxo.frozen = now_frozen
                    break
        return now_frozen

    def is_utxo_frozen(self, outpoint: str) -> bool:
        """Check if a UTXO is frozen.

        Args:
            outpoint: Outpoint string in ``txid:vout`` format.

        Returns:
            True if frozen, False otherwise.
        """
        if self.metadata_store is None:
            return False
        return self.metadata_store.is_frozen(outpoint)

Attributes

address_cache: dict[str, tuple[int, int, int]] = {} instance-attribute

addresses_with_history: set[str] = set() instance-attribute

backend = backend instance-attribute

data_dir = data_dir instance-attribute

fidelity_bond_locktime_cache: dict[str, int] = {} instance-attribute

gap_limit = gap_limit instance-attribute

issued_receive_addresses: set[str] = set() instance-attribute

master_key = HDKey.from_seed(seed) instance-attribute

max_sats_freeze_reuse = max_sats_freeze_reuse instance-attribute

metadata_store: UTXOMetadataStore | None = None instance-attribute

mixdepth_count = mixdepth_count instance-attribute

network = network instance-attribute

reconstruct_history_enabled: bool = reconstruct_history instance-attribute

reserved_address_labels: dict[str, str] = {} instance-attribute

reserved_addresses: set[str] = set() instance-attribute

root_path = f'm/84'/{coin_type}'' instance-attribute

scan_range = scan_range instance-attribute

utxo_cache: dict[int, list[UTXOInfo]] = {} instance-attribute

wallet_fingerprint = fingerprint instance-attribute

Methods:

__init__(mnemonic: str, backend: BlockchainBackend, network: str = 'mainnet', mixdepth_count: int = 5, gap_limit: int = 20, scan_range: int = DEFAULT_SCAN_RANGE, data_dir: Path | None = None, passphrase: str = '', max_sats_freeze_reuse: int = -1, reconstruct_history: bool = True)

Source code in jmwallet/src/jmwallet/wallet/service.py

def __init__(
    self,
    mnemonic: str,
    backend: BlockchainBackend,
    network: str = "mainnet",
    mixdepth_count: int = 5,
    gap_limit: int = 20,
    scan_range: int = DEFAULT_SCAN_RANGE,
    data_dir: Path | None = None,
    passphrase: str = "",
    max_sats_freeze_reuse: int = -1,
    reconstruct_history: bool = True,
):
    self.backend = backend
    self.network = network
    self.mixdepth_count = mixdepth_count
    # ``gap_limit`` is the BIP44 trailing-empty threshold (default 20).
    # ``scan_range`` is the descriptor lookahead window imported into
    # Bitcoin Core (default 1000). The two concepts used to be conflated
    # via a ``max(1000, gap_limit * 10)`` formula, dropped in favor of
    # explicit configuration (issue #475).
    self.gap_limit = gap_limit
    self.scan_range = scan_range
    self.data_dir = data_dir
    # Forced address-reuse defense (issue #529): a UTXO that lands on an
    # already-used wallet address is auto-frozen during sync when its value
    # is <= ``max_sats_freeze_reuse`` (or always, when it is -1). 0 disables
    # the behavior. Matches legacy joinmarket-clientserver's
    # ``POLICY.max_sats_freeze_reuse``.
    self.max_sats_freeze_reuse = max_sats_freeze_reuse

    seed = mnemonic_to_seed(mnemonic, passphrase)
    self.master_key = HDKey.from_seed(seed)

    coin_type = 0 if network == "mainnet" else 1
    self.root_path = f"m/84'/{coin_type}'"

    # Log fingerprint for debugging (helps identify passphrase issues)
    fingerprint = self.master_key.derive("m/0").fingerprint.hex()
    # Expose fingerprint as a stable wallet identifier (issue #473).
    # This is the same 8-char hex used by the descriptor wallet name and
    # by the CoinJoin history CSV to scope entries to a specific wallet.
    self.wallet_fingerprint = fingerprint
    logger.info(
        f"Initialized wallet: fingerprint={fingerprint}, "
        f"mixdepths={mixdepth_count}, network={network}, "
        f"passphrase={'(set)' if passphrase else '(none)'}"
    )

    self.address_cache: dict[str, tuple[int, int, int]] = {}
    self._path_cache: dict[tuple[int, int, int], str] = {}
    self.utxo_cache: dict[int, list[UTXOInfo]] = {}
    # Forced-address-reuse defense state (issue #529, hardened for #542).
    #
    # The auto-freeze must distinguish a *genuine* forced reuse (a new coin
    # landing on an address we funded and then emptied) from a perfectly
    # legitimate first-use coin that merely became visible on a later sync
    # (background descriptor rescan still catching up, a transient RPC
    # failure on the first sync, or a descriptor-range upgrade). The
    # persistent ``addresses_with_history`` set is restored at init, so it
    # cannot be used on its own to decide "this address was emptied": a
    # still-funded first-use address is in that set too. Relying on it (via
    # the old one-shot ``_just_initialized`` guard) wrongly froze coins that
    # the first sync had not yet observed (#542).
    #
    # Instead we accumulate the addresses and outpoints we have *positively
    # observed funded*. A coin is treated as forced reuse only when its
    # address was seen funded earlier and is empty again now (see
    # :meth:`_auto_freeze_reused_address_utxos`). These sets are persisted
    # in the metadata store and reseeded below, so the knowledge survives
    # restarts: an address emptied before a restart and refunded after it is
    # still frozen, while coins that predate the restart (in the persisted
    # seen-outpoint set) and late-discovered first-use coins (never
    # persisted as observed-funded) are left spendable.
    self._observed_funded_addresses: set[str] = set()
    self._observed_outpoints: set[str] = set()
    # Guards the once-per-process import-label reconstruction pass (see
    # WalletSyncMixin.reconstruct_imported_labels). Coins received while
    # running are either this wallet's own CoinJoins (recorded in history)
    # or genuine deposits, so only the imported backlog needs scanning.
    self._imported_labels_scanned: bool = False
    # Guards the once-per-process import-history reconstruction pass (see
    # WalletSyncMixin.reconstruct_imported_history). The automatic pass
    # only fires for wallets with no recorded history (seed imports);
    # ``reconstruct_history`` maps the [wallet] reconstruct_history config
    # toggle.
    self._imported_history_scanned: bool = False
    # True once an empty-history wallet has entered the imported-history
    # workflow, even if reconstruction was deferred for a Core rescan.
    # This keeps a protocol row written during that rescan from cancelling
    # the pending backfill later in the same process.
    self._imported_history_started: bool = False
    self.reconstruct_history_enabled: bool = reconstruct_history

    # UTXO + address metadata store (BIP-329 JSONL). Frozen UTXO state,
    # output labels, and the persistent "addresses with on-chain history"
    # set all live in the same per-wallet ``wallet_metadata_<fp>.jsonl``
    # file. Partitioning by fingerprint is mandatory: pre-0.30.0 builds
    # used a shared ``wallet_metadata.jsonl`` per data_dir, which leaked
    # one wallet's used-address set and frozen-UTXO state into any
    # other wallet opened in the same directory.
    self.metadata_store: UTXOMetadataStore | None = None
    if data_dir is not None:
        # Only pre-derive owned addresses when the one-shot migration
        # from the legacy shared file is actually going to run. After
        # the first open, the per-wallet file exists and the
        # migration is skipped, so the typical hot path pays nothing.
        from jmcore.paths import get_wallet_metadata_path

        per_wallet_path = get_wallet_metadata_path(data_dir, fingerprint=fingerprint)
        shared_path = get_wallet_metadata_path(data_dir, fingerprint=None)
        owned_addresses: set[str] | None = None
        if not per_wallet_path.exists() and shared_path.exists():
            # Derivation is pure compute (no backend RPCs); for the
            # default scan_range=1000 / mixdepth_count=5 this is
            # roughly 10k BIP32 derivations and completes in well
            # under a second. The derived addresses are also seeded
            # into ``self.address_cache`` so subsequent lookups
            # skip re-derivation.
            owned_addresses = set()
            for mixdepth in range(self.mixdepth_count):
                for change in (0, 1):
                    for index in range(self.scan_range):
                        owned_addresses.add(self.get_address(mixdepth, change, index))
        self.metadata_store = load_metadata_store(
            data_dir,
            fingerprint=fingerprint,
            owned_addresses=owned_addresses,
        )

    # Track addresses that have ever had UTXOs (including spent ones).
    # Used to label addresses as "used-empty" vs "new" and, critically,
    # to prevent reissuing a previously-funded deposit address. Backed by
    # the metadata store so the knowledge survives across runs (light
    # clients like Neutrino and Bitcoin Core's address-book-bound
    # ``listreceivedbyaddress`` cannot always rediscover spent-then-empty
    # addresses from scratch).
    self.addresses_with_history: set[str] = set()
    if self.metadata_store is not None:
        self.addresses_with_history.update(self.metadata_store.get_used_addresses())
        self._migrate_legacy_address_history(data_dir)
        # Reseed the forced-address-reuse observation sets so the defense
        # survives restarts (issue #559).
        self._observed_funded_addresses.update(
            self.metadata_store.get_observed_funded_addresses()
        )
        self._observed_outpoints.update(self.metadata_store.get_seen_outpoints())

    # Track addresses currently reserved for in-progress CoinJoin sessions
    # These addresses have been shared with a taker but the CoinJoin hasn't
    # completed yet. They must not be reused until the session ends.
    self.reserved_addresses: set[str] = set()
    # Track receive addresses that were already handed out via API/CLI.
    # Even if they do not appear on-chain yet, we should not reissue them.
    self.issued_receive_addresses: set[str] = set()
    # Reserved (set-aside) deposit addresses -> optional user label.
    # Persisted in the metadata store (``jm:reserved`` records) so handed-out
    # and user-labeled deposit addresses survive restarts, are never
    # reissued, and can be shown with their label in the extended view.
    self.reserved_address_labels: dict[str, str] = {}
    if self.metadata_store is not None:
        self.reserved_address_labels.update(self.metadata_store.get_reserved_labels())
        # Feed the in-memory picker skip-set so reserved addresses are not
        # reissued as the next unused deposit address after a restart.
        self.issued_receive_addresses.update(self.reserved_address_labels.keys())
    # Cache for fidelity bond locktimes (address -> locktime)
    self.fidelity_bond_locktime_cache: dict[str, int] = {}
    # Lazily-built cache of every canonical fidelity-bond address (all
    # 960 timenumbers) mapped to its (locktime, timenumber). Populated on
    # first use by ``WalletSyncMixin._canonical_bond_address_map``; see
    # that method for why this exists (recognizing bond UTXOs Bitcoin
    # Core already tracks even when the local registry has no matching
    # entry, issue: fidelity bonds invisible after per-wallet registry
    # partition / #492 migration gaps).
    self._canonical_bond_addresses: dict[str, tuple[int, int]] | None = None

    # One-shot migration of the legacy shared ``fidelity_bonds.json``
    # registry into a per-wallet ``fidelity_bonds_<fp>.json`` file
    # (issue #492). Same shape as the metadata-store migration above:
    # only runs when the per-wallet file is missing AND the legacy
    # file exists, so the typical hot path is a single ``Path.exists``
    # check.
    if data_dir is not None:
        self._migrate_legacy_bond_registry(data_dir)

    # Resolve reserved deposit addresses to their derivation path and seed
    # the address cache. The deposit-address pickers key on the cache to
    # map an address to its index, so this lets them advance past a
    # reserved address even before a full sync has populated the cache
    # (after a restart the reservation is loaded from disk, but the
    # address itself has not been derived yet). Reserved addresses are few
    # and typically at low indices, so this is cheap.
    for reserved_addr in list(self.reserved_address_labels):
        if reserved_addr not in self.address_cache:
            try:
                self._find_address_path(reserved_addr, max_scan=self.scan_range)
            except Exception as exc:  # pragma: no cover - defensive
                logger.debug(f"Could not resolve reserved address {reserved_addr}: {exc}")

close() -> None async

Close backend connection

Source code in jmwallet/src/jmwallet/wallet/service.py

async def close(self) -> None:
    """Close backend connection"""
    await self.backend.close()

find_utxo_by_address(address: str) -> UTXOInfo | None

Find a UTXO by its address across all mixdepths.

This is useful for matching CoinJoin outputs to history entries. Returns the first matching UTXO found, or None if address not found.

Args: address: Bitcoin address to search for

Returns: UTXOInfo if found, None otherwise

Source code in jmwallet/src/jmwallet/wallet/service.py

def find_utxo_by_address(self, address: str) -> UTXOInfo | None:
    """
    Find a UTXO by its address across all mixdepths.

    This is useful for matching CoinJoin outputs to history entries.
    Returns the first matching UTXO found, or None if address not found.

    Args:
        address: Bitcoin address to search for

    Returns:
        UTXOInfo if found, None otherwise
    """
    for mixdepth in range(self.mixdepth_count):
        utxos = self.utxo_cache.get(mixdepth, [])
        for utxo in utxos:
            if utxo.address == address:
                return utxo
    return None

freeze_utxo(outpoint: str) -> None

Freeze a UTXO by outpoint (persisted to disk).

Args: outpoint: Outpoint string in txid:vout format.

Raises: RuntimeError: If no metadata store is available (no data_dir).

Source code in jmwallet/src/jmwallet/wallet/service.py

def freeze_utxo(self, outpoint: str) -> None:
    """Freeze a UTXO by outpoint (persisted to disk).

    Args:
        outpoint: Outpoint string in ``txid:vout`` format.

    Raises:
        RuntimeError: If no metadata store is available (no data_dir).
    """
    if self.metadata_store is None:
        raise RuntimeError("Cannot freeze UTXOs without a data directory")
    self.metadata_store.freeze(outpoint)
    # Update the in-memory UTXO cache
    for utxos in self.utxo_cache.values():
        for utxo in utxos:
            if utxo.outpoint == outpoint:
                utxo.frozen = True
                return

get_account_xpub(mixdepth: int) -> str

Get the extended public key (xpub) for a mixdepth account.

Derives the key at path m/84'/coin'/mixdepth' and returns its xpub. This xpub can be used in Bitcoin Core descriptors for efficient scanning.

Args: mixdepth: The mixdepth (account) number (0-4)

Returns: xpub/tpub string for the account

Source code in jmwallet/src/jmwallet/wallet/service.py

def get_account_xpub(self, mixdepth: int) -> str:
    """
    Get the extended public key (xpub) for a mixdepth account.

    Derives the key at path m/84'/coin'/mixdepth' and returns its xpub.
    This xpub can be used in Bitcoin Core descriptors for efficient scanning.

    Args:
        mixdepth: The mixdepth (account) number (0-4)

    Returns:
        xpub/tpub string for the account
    """
    account_path = f"{self.root_path}/{mixdepth}'"
    account_key = self.master_key.derive(account_path)
    return account_key.get_xpub(self.network)

get_account_zpub(mixdepth: int) -> str

Get the BIP84 extended public key (zpub) for a mixdepth account.

Derives the key at path m/84'/coin'/mixdepth' and returns its zpub. zpub explicitly indicates this is a native segwit (P2WPKH) wallet.

Args: mixdepth: The mixdepth (account) number (0-4)

Returns: zpub/vpub string for the account

Source code in jmwallet/src/jmwallet/wallet/service.py

def get_account_zpub(self, mixdepth: int) -> str:
    """
    Get the BIP84 extended public key (zpub) for a mixdepth account.

    Derives the key at path m/84'/coin'/mixdepth' and returns its zpub.
    zpub explicitly indicates this is a native segwit (P2WPKH) wallet.

    Args:
        mixdepth: The mixdepth (account) number (0-4)

    Returns:
        zpub/vpub string for the account
    """
    account_path = f"{self.root_path}/{mixdepth}'"
    account_key = self.master_key.derive(account_path)
    return account_key.get_zpub(self.network)

get_address(mixdepth: int, change: int, index: int) -> str

Get address for given path

Source code in jmwallet/src/jmwallet/wallet/service.py

def get_address(self, mixdepth: int, change: int, index: int) -> str:
    """Get address for given path"""
    if mixdepth >= self.mixdepth_count:
        raise ValueError(f"Mixdepth {mixdepth} exceeds maximum {self.mixdepth_count}")

    path_key = (mixdepth, change, index)
    cached = self._path_cache.get(path_key)
    if cached is not None:
        self.address_cache[cached] = path_key
        return cached

    path = f"{self.root_path}/{mixdepth}'/{change}/{index}"
    key = self.master_key.derive(path)
    address = key.get_address(self.network)

    self.address_cache[address] = (mixdepth, change, index)
    self._path_cache[path_key] = address

    return address

get_balance(mixdepth: int, include_fidelity_bonds: bool = True, min_confirmations: int = 0) -> int async

Get balance for a mixdepth.

Args: mixdepth: Mixdepth to get balance for include_fidelity_bonds: If True (default), include fidelity bond UTXOs. If False, exclude fidelity bond UTXOs. min_confirmations: Minimum confirmations required (default: 0).

Note: Frozen UTXOs are excluded from balance calculations.

Source code in jmwallet/src/jmwallet/wallet/service.py

async def get_balance(
    self, mixdepth: int, include_fidelity_bonds: bool = True, min_confirmations: int = 0
) -> int:
    """Get balance for a mixdepth.

    Args:
        mixdepth: Mixdepth to get balance for
        include_fidelity_bonds: If True (default), include fidelity bond UTXOs.
                                If False, exclude fidelity bond UTXOs.
        min_confirmations: Minimum confirmations required (default: 0).

    Note:
        Frozen UTXOs are excluded from balance calculations.
    """
    if mixdepth not in self.utxo_cache:
        await self.sync_mixdepth(mixdepth)

    utxos = self.utxo_cache.get(mixdepth, [])
    utxos = [u for u in utxos if not u.frozen]
    if not include_fidelity_bonds:
        utxos = [u for u in utxos if not u.is_fidelity_bond]
    if min_confirmations > 0:
        utxos = [u for u in utxos if u.confirmations >= min_confirmations]
    return sum(utxo.value for utxo in utxos)

get_balance_for_offers(mixdepth: int, min_confirmations: int = 0, *, restrict_md0: bool = True) -> int async

Get balance available for maker offers (excludes fidelity bond UTXOs).

Fidelity bonds should never be automatically spent in CoinJoins, so makers must exclude them when calculating available offer amounts.

For mixdepth 0 (when restrict_md0 is True), UTXOs that are not CoinJoin outputs are restricted to a single UTXO to avoid linking deposits or fidelity bonds. CoinJoin outputs (label == "cj-out") are exempt because they already have CoinJoin privacy and can be safely merged.

The effective balance is therefore::

max(sum_of_cj_outputs, largest_non_cj_output)

When restrict_md0 is False (opt-in via config), mixdepth 0 is treated the same as any other mixdepth.

Source code in jmwallet/src/jmwallet/wallet/service.py

async def get_balance_for_offers(
    self, mixdepth: int, min_confirmations: int = 0, *, restrict_md0: bool = True
) -> int:
    """Get balance available for maker offers (excludes fidelity bond UTXOs).

    Fidelity bonds should never be automatically spent in CoinJoins,
    so makers must exclude them when calculating available offer amounts.

    For mixdepth 0 (when ``restrict_md0`` is True), UTXOs that are **not**
    CoinJoin outputs are restricted to a single UTXO to avoid linking
    deposits or fidelity bonds.  CoinJoin outputs (``label == "cj-out"``)
    are exempt because they already have CoinJoin privacy and can be
    safely merged.

    The effective balance is therefore::

        max(sum_of_cj_outputs, largest_non_cj_output)

    When ``restrict_md0`` is False (opt-in via config), mixdepth 0 is
    treated the same as any other mixdepth.
    """
    if mixdepth == 0 and restrict_md0:
        if mixdepth not in self.utxo_cache:
            await self.sync_mixdepth(mixdepth)
        utxos = self.utxo_cache.get(mixdepth, [])
        eligible = [
            u
            for u in utxos
            if not u.frozen and not u.is_fidelity_bond and u.confirmations >= min_confirmations
        ]
        if not eligible:
            return 0

        cj_pool = sum(u.value for u in eligible if u.label == "cj-out")
        non_cj = [u for u in eligible if u.label != "cj-out"]
        largest_single = max((u.value for u in non_cj), default=0)
        return max(cj_pool, largest_single)

    return await self.get_balance(
        mixdepth, include_fidelity_bonds=False, min_confirmations=min_confirmations
    )

get_change_address(mixdepth: int, index: int) -> str

Get internal (change) address

Source code in jmwallet/src/jmwallet/wallet/service.py

def get_change_address(self, mixdepth: int, index: int) -> str:
    """Get internal (change) address"""
    return self.get_address(mixdepth, 1, index)

get_fidelity_bond_address(index: int, locktime: int) -> str

Get a fidelity bond P2WSH address.

Creates a timelocked script: OP_CLTV OP_DROP OP_CHECKSIG wrapped in P2WSH.

The index parameter is a legacy argument and is ignored; the BIP32 child index is always the timenumber derived from locktime.

Args: index: Legacy address index (ignored; timenumber is used instead) locktime: Unix timestamp for the timelock

Returns: P2WSH address for the fidelity bond

Source code in jmwallet/src/jmwallet/wallet/service.py

def get_fidelity_bond_address(self, index: int, locktime: int) -> str:
    """
    Get a fidelity bond P2WSH address.

    Creates a timelocked script: <locktime> OP_CLTV OP_DROP <pubkey> OP_CHECKSIG
    wrapped in P2WSH.

    The ``index`` parameter is a legacy argument and is **ignored**; the
    BIP32 child index is always the timenumber derived from ``locktime``.

    Args:
        index: Legacy address index (ignored; timenumber is used instead)
        locktime: Unix timestamp for the timelock

    Returns:
        P2WSH address for the fidelity bond
    """
    from jmcore.timenumber import timestamp_to_timenumber

    key = self.get_fidelity_bond_key(index, locktime)
    pubkey_hex = key.get_public_key_bytes(compressed=True).hex()

    # Create the timelock script
    script = mk_freeze_script(pubkey_hex, locktime)

    # Convert to P2WSH address
    address = script_to_p2wsh_address(script, self.network)

    # Cache with timenumber as the index (matches BIP32 child index)
    timenumber = timestamp_to_timenumber(locktime)
    self.address_cache[address] = (0, FIDELITY_BOND_BRANCH, timenumber)
    # Also store the locktime in a separate cache for fidelity bonds
    self.fidelity_bond_locktime_cache[address] = locktime

    logger.trace(f"Created fidelity bond address {address} with locktime {locktime}")
    return address

get_fidelity_bond_balance(mixdepth: int) -> int async

Get balance of fidelity bond UTXOs for a mixdepth.

Note: Unlike spendable-balance helpers, the frozen flag is not applied here. A fidelity bond is already excluded from automatic coin selection by virtue of being a timelocked bond, so its frozen flag is orthogonal to its informational value. The maker advertises the bond, list-bonds reports it as ACTIVE, and the extended wallet view counts it regardless of frozen; this helper backs the basic jm-wallet info (+... FB) annotation and must report the same bond value so the views stay consistent (see issue: bond hidden from basic info after freeze).

Source code in jmwallet/src/jmwallet/wallet/service.py

async def get_fidelity_bond_balance(self, mixdepth: int) -> int:
    """Get balance of fidelity bond UTXOs for a mixdepth.

    Note:
        Unlike spendable-balance helpers, the ``frozen`` flag is **not**
        applied here. A fidelity bond is already excluded from automatic
        coin selection by virtue of being a timelocked bond, so its
        ``frozen`` flag is orthogonal to its informational value. The
        maker advertises the bond, ``list-bonds`` reports it as ACTIVE,
        and the extended wallet view counts it regardless of ``frozen``;
        this helper backs the basic ``jm-wallet info`` ``(+... FB)``
        annotation and must report the same bond value so the views stay
        consistent (see issue: bond hidden from basic info after freeze).
    """
    if mixdepth not in self.utxo_cache:
        await self.sync_mixdepth(mixdepth)

    utxos = self.utxo_cache.get(mixdepth, [])
    return sum(utxo.value for utxo in utxos if utxo.is_fidelity_bond)

get_fidelity_bond_key(index: int, locktime: int) -> HDKey

Get the HD key for a fidelity bond.

Fidelity bond path: m/84'/coin'/0'/2/timenumber

In the JoinMarket protocol, the BIP32 child index for fidelity bonds is the timenumber (0-959), NOT a separate address index. Each timenumber maps 1:1 to a locktime (1st of month, Jan 2020 - Dec 2099).

For backward compatibility, the index parameter is still accepted but is ignored when locktime is a valid timenumber locktime. The timenumber is computed from the locktime and used as the child index.

Args: index: Legacy address index (ignored when locktime is valid). Kept for API compatibility. locktime: Unix timestamp for the timelock. Must be a valid timenumber locktime (1st of month, midnight UTC).

Returns: HDKey for the fidelity bond

Source code in jmwallet/src/jmwallet/wallet/service.py

def get_fidelity_bond_key(self, index: int, locktime: int) -> HDKey:
    """
    Get the HD key for a fidelity bond.

    Fidelity bond path: m/84'/coin'/0'/2/timenumber

    In the JoinMarket protocol, the BIP32 child index for fidelity bonds
    is the **timenumber** (0-959), NOT a separate address index. Each
    timenumber maps 1:1 to a locktime (1st of month, Jan 2020 - Dec 2099).

    For backward compatibility, the ``index`` parameter is still accepted
    but is **ignored** when ``locktime`` is a valid timenumber locktime.
    The timenumber is computed from the locktime and used as the child index.

    Args:
        index: Legacy address index (ignored when locktime is valid).
               Kept for API compatibility.
        locktime: Unix timestamp for the timelock. Must be a valid
                  timenumber locktime (1st of month, midnight UTC).

    Returns:
        HDKey for the fidelity bond
    """
    from jmcore.timenumber import timestamp_to_timenumber

    # The BIP32 child index is the timenumber derived from the locktime,
    # matching the reference JoinMarket implementation.
    timenumber = timestamp_to_timenumber(locktime)
    path = f"{self.root_path}/0'/{FIDELITY_BOND_BRANCH}/{timenumber}"
    return self.master_key.derive(path)

get_fidelity_bond_script(index: int, locktime: int) -> bytes

Get the redeem script for a fidelity bond.

The index parameter is a legacy argument and is ignored; the BIP32 child index is always the timenumber derived from locktime.

Args: index: Legacy address index (ignored; timenumber is used instead) locktime: Unix timestamp for the timelock

Returns: Timelock redeem script bytes

Source code in jmwallet/src/jmwallet/wallet/service.py

def get_fidelity_bond_script(self, index: int, locktime: int) -> bytes:
    """
    Get the redeem script for a fidelity bond.

    The ``index`` parameter is a legacy argument and is **ignored**; the
    BIP32 child index is always the timenumber derived from ``locktime``.

    Args:
        index: Legacy address index (ignored; timenumber is used instead)
        locktime: Unix timestamp for the timelock

    Returns:
        Timelock redeem script bytes
    """
    key = self.get_fidelity_bond_key(index, locktime)
    pubkey_hex = key.get_public_key_bytes(compressed=True).hex()
    return mk_freeze_script(pubkey_hex, locktime)

get_key_for_address(address: str) -> HDKey | None

Get HD key for a known address

Source code in jmwallet/src/jmwallet/wallet/service.py

def get_key_for_address(self, address: str) -> HDKey | None:
    """Get HD key for a known address"""
    path_info = self.address_cache.get(address)
    if path_info is None:
        path_info = self.address_cache.get(address.lower())
    if path_info is None:
        return None

    mixdepth, change, index = path_info
    path = f"{self.root_path}/{mixdepth}'/{change}/{index}"
    return self.master_key.derive(path)

get_locked_input_outpoints() -> set[tuple[str, int]]

Return (txid, vout) inputs currently locked by any in-flight round.

Re-reads the on-disk metadata so locks written by other processes (another taker round, or a maker serving a different taker) are visible right before coin selection. Returns an empty set when no metadata store is configured (no data directory).

Source code in jmwallet/src/jmwallet/wallet/service.py

def get_locked_input_outpoints(self) -> set[tuple[str, int]]:
    """Return ``(txid, vout)`` inputs currently locked by any in-flight round.

    Re-reads the on-disk metadata so locks written by other processes
    (another taker round, or a maker serving a different taker) are visible
    right before coin selection. Returns an empty set when no metadata store
    is configured (no data directory).
    """
    if self.metadata_store is None:
        return set()
    self.metadata_store.load()
    locked: set[tuple[str, int]] = set()
    for ref in self.metadata_store.get_locked_outpoints():
        txid, _, vout = ref.rpartition(":")
        if txid and vout.isdigit():
            locked.add((txid, int(vout)))
    return locked

get_locktime_for_address(address: str) -> int | None

Get the locktime for a fidelity bond address.

Args: address: The fidelity bond address

Returns: Locktime as Unix timestamp, or None if not a fidelity bond address

Source code in jmwallet/src/jmwallet/wallet/service.py

def get_locktime_for_address(self, address: str) -> int | None:
    """
    Get the locktime for a fidelity bond address.

    Args:
        address: The fidelity bond address

    Returns:
        Locktime as Unix timestamp, or None if not a fidelity bond address
    """
    return self.fidelity_bond_locktime_cache.get(address)

get_new_address(mixdepth: int) -> str

Get next unused receive address for a mixdepth.

Synchronous fast path: returns the address at get_next_address_index(mixdepth, 0). This relies on the sync-layer addresses_with_history being complete; if the last bulk enumeration was truncated by an RPC failure, this method may return a previously-funded address.

Privacy-critical callers (CLI info / daemon address endpoints) should prefer :meth:get_new_address_verified, which adds a per-candidate getreceivedbyaddress check.

Source code in jmwallet/src/jmwallet/wallet/service.py

def get_new_address(self, mixdepth: int) -> str:
    """Get next unused receive address for a mixdepth.

    Synchronous fast path: returns the address at
    ``get_next_address_index(mixdepth, 0)``. This relies on the
    sync-layer ``addresses_with_history`` being complete; if the
    last bulk enumeration was truncated by an RPC failure, this
    method may return a previously-funded address.

    Privacy-critical callers (CLI ``info`` / daemon address
    endpoints) should prefer :meth:`get_new_address_verified`,
    which adds a per-candidate ``getreceivedbyaddress`` check.
    """
    next_index = self.get_next_address_index(mixdepth, 0)
    address = self.get_receive_address(mixdepth, next_index)
    self.reserve_address(address)
    return address

get_new_address_verified(mixdepth: int) -> str async

Async deposit-address picker with on-chain verification.

Wraps :meth:get_next_safe_deposit_address and reserves the chosen address (persisted when a data_dir is configured) so it is never reissued, even across restarts. Use this from any async code path that exposes a deposit address to users or peers (jm-wallet info, jmwalletd /wallet/address/new, maker/taker deposit prompts).

Source code in jmwallet/src/jmwallet/wallet/service.py

async def get_new_address_verified(self, mixdepth: int) -> str:
    """Async deposit-address picker with on-chain verification.

    Wraps :meth:`get_next_safe_deposit_address` and reserves the chosen
    address (persisted when a ``data_dir`` is configured) so it is never
    reissued, even across restarts. Use this from any async code path that
    exposes a deposit address to users or peers (``jm-wallet info``,
    jmwalletd ``/wallet/address/new``, maker/taker deposit prompts).
    """
    address, _ = await self.get_next_safe_deposit_address(mixdepth)
    self.reserve_address(address)
    return address

get_next_address_index(mixdepth: int, change: int) -> int

Get next unused address index for mixdepth/change.

Returns the highest index + 1 among all addresses that have ever been used, ensuring we never reuse addresses. An address is considered "used" if it: - Has current UTXOs - Had UTXOs in the past (tracked in addresses_with_history) - Appears in CoinJoin history (even if never funded)

We always return one past the highest used index, even if lower indices appear unused. Those may have been skipped for a reason (e.g., shared in a failed CoinJoin, or spent in an internal transfer).

Source code in jmwallet/src/jmwallet/wallet/service.py

def get_next_address_index(self, mixdepth: int, change: int) -> int:
    """
    Get next unused address index for mixdepth/change.

    Returns the highest index + 1 among all addresses that have ever been used,
    ensuring we never reuse addresses. An address is considered "used" if it:
    - Has current UTXOs
    - Had UTXOs in the past (tracked in addresses_with_history)
    - Appears in CoinJoin history (even if never funded)

    We always return one past the highest used index, even if lower indices
    appear unused. Those may have been skipped for a reason (e.g., shared in
    a failed CoinJoin, or spent in an internal transfer).
    """
    max_index = -1

    # Check addresses with current UTXOs
    utxos = self.utxo_cache.get(mixdepth, [])
    for utxo in utxos:
        if utxo.address in self.address_cache:
            md, ch, idx = self.address_cache[utxo.address]
            if md == mixdepth and ch == change and idx > max_index:
                max_index = idx

    # Check addresses that ever had blockchain activity (including spent)
    for address in self.addresses_with_history:
        if address in self.address_cache:
            md, ch, idx = self.address_cache[address]
            if md == mixdepth and ch == change and idx > max_index:
                max_index = idx

    # Check CoinJoin history for addresses that may have been shared
    # but never received funds (e.g., failed CoinJoins)
    if self.data_dir:
        from jmwallet.history import get_used_addresses

        cj_addresses = get_used_addresses(
            self.data_dir, wallet_fingerprint=self.wallet_fingerprint
        )
        self._prune_reserved_addresses(cj_addresses | self.addresses_with_history)
        for address in cj_addresses:
            if address in self.address_cache:
                md, ch, idx = self.address_cache[address]
                if md == mixdepth and ch == change and idx > max_index:
                    max_index = idx

    # Check addresses reserved for in-progress CoinJoin sessions
    # These have been shared with takers but the session hasn't completed yet
    for address in self.reserved_addresses:
        if address in self.address_cache:
            md, ch, idx = self.address_cache[address]
            if md == mixdepth and ch == change and idx > max_index:
                max_index = idx

    # Check receive addresses that were already issued to callers.
    # This prevents repeated GET /address/new/{mixdepth} calls from
    # returning the same address when no on-chain history exists yet.
    for address in self.issued_receive_addresses:
        if address in self.address_cache:
            md, ch, idx = self.address_cache[address]
            if md == mixdepth and ch == change and idx > max_index:
                max_index = idx

    return max_index + 1

get_private_key(mixdepth: int, change: int, index: int) -> bytes

Get private key for given path

Source code in jmwallet/src/jmwallet/wallet/service.py

def get_private_key(self, mixdepth: int, change: int, index: int) -> bytes:
    """Get private key for given path"""
    path = f"{self.root_path}/{mixdepth}'/{change}/{index}"
    key = self.master_key.derive(path)
    return key.get_private_key_bytes()

get_receive_address(mixdepth: int, index: int) -> str

Get external (receive) address

Source code in jmwallet/src/jmwallet/wallet/service.py

def get_receive_address(self, mixdepth: int, index: int) -> str:
    """Get external (receive) address"""
    return self.get_address(mixdepth, 0, index)

get_reserved_addresses() -> dict[str, str]

Return a copy of the reserved address -> user label mapping.

Source code in jmwallet/src/jmwallet/wallet/service.py

def get_reserved_addresses(self) -> dict[str, str]:
    """Return a copy of the reserved address -> user label mapping."""
    return dict(self.reserved_address_labels)

get_scan_descriptors(scan_range: int = DEFAULT_SCAN_RANGE) -> list[dict[str, Any]]

Generate descriptors for efficient UTXO scanning with Bitcoin Core.

Creates wpkh() descriptors with xpub and range for all mixdepths, both external (receive) and internal (change) addresses.

Using descriptors with ranges is much more efficient than scanning individual addresses, as Bitcoin Core can scan the entire range in a single pass through the UTXO set.

Args: scan_range: Maximum index to scan (default 1000, Bitcoin Core's default)

Returns: List of descriptor dicts for use with scantxoutset: [{"desc": "wpkh(xpub.../0/*)", "range": [0, 999]}, ...]

Source code in jmwallet/src/jmwallet/wallet/service.py

def get_scan_descriptors(self, scan_range: int = DEFAULT_SCAN_RANGE) -> list[dict[str, Any]]:
    """
    Generate descriptors for efficient UTXO scanning with Bitcoin Core.

    Creates wpkh() descriptors with xpub and range for all mixdepths,
    both external (receive) and internal (change) addresses.

    Using descriptors with ranges is much more efficient than scanning
    individual addresses, as Bitcoin Core can scan the entire range in
    a single pass through the UTXO set.

    Args:
        scan_range: Maximum index to scan (default 1000, Bitcoin Core's default)

    Returns:
        List of descriptor dicts for use with scantxoutset:
        [{"desc": "wpkh(xpub.../0/*)", "range": [0, 999]}, ...]
    """
    descriptors = []

    for mixdepth in range(self.mixdepth_count):
        xpub = self.get_account_xpub(mixdepth)

        # External (receive) addresses: .../0/*
        descriptors.append({"desc": f"wpkh({xpub}/0/*)", "range": [0, scan_range - 1]})

        # Internal (change) addresses: .../1/*
        descriptors.append({"desc": f"wpkh({xpub}/1/*)", "range": [0, scan_range - 1]})

    logger.debug(
        f"Generated {len(descriptors)} descriptors for {self.mixdepth_count} mixdepths "
        f"with range [0, {scan_range - 1}]"
    )
    return descriptors

get_total_balance(include_fidelity_bonds: bool = True, min_confirmations: int = 0) -> int async

Get the spendable balance across all mixdepths.

Despite the name, this is the spendable total: frozen UTXOs are always excluded, and fidelity bonds are excluded when include_fidelity_bonds is False. Callers that need the grand total (including frozen funds) must add the frozen amount back themselves.

Args: include_fidelity_bonds: If True (default), include fidelity bond UTXOs. If False, exclude fidelity bond UTXOs. min_confirmations: Minimum confirmations required (default: 0).

Note: Frozen UTXOs are excluded from balance calculations.

Source code in jmwallet/src/jmwallet/wallet/service.py

async def get_total_balance(
    self, include_fidelity_bonds: bool = True, min_confirmations: int = 0
) -> int:
    """Get the spendable balance across all mixdepths.

    Despite the name, this is the *spendable* total: frozen UTXOs are
    always excluded, and fidelity bonds are excluded when
    ``include_fidelity_bonds`` is False. Callers that need the grand total
    (including frozen funds) must add the frozen amount back themselves.

    Args:
        include_fidelity_bonds: If True (default), include fidelity bond UTXOs.
                                If False, exclude fidelity bond UTXOs.
        min_confirmations: Minimum confirmations required (default: 0).

    Note:
        Frozen UTXOs are excluded from balance calculations.
    """
    total = 0
    for mixdepth in range(self.mixdepth_count):
        balance = await self.get_balance(
            mixdepth,
            include_fidelity_bonds=include_fidelity_bonds,
            min_confirmations=min_confirmations,
        )
        total += balance
    return total

get_utxos(mixdepth: int) -> list[UTXOInfo] async

Get UTXOs for a mixdepth, syncing if not cached.

Source code in jmwallet/src/jmwallet/wallet/service.py

async def get_utxos(self, mixdepth: int) -> list[UTXOInfo]:
    """Get UTXOs for a mixdepth, syncing if not cached."""
    if mixdepth not in self.utxo_cache:
        await self.sync_mixdepth(mixdepth)
    return self.utxo_cache.get(mixdepth, [])

is_address_reserved(address: str) -> bool

Return True if address has been reserved/set aside by the user.

Source code in jmwallet/src/jmwallet/wallet/service.py

def is_address_reserved(self, address: str) -> bool:
    """Return True if ``address`` has been reserved/set aside by the user."""
    return address in self.reserved_address_labels

is_utxo_frozen(outpoint: str) -> bool

Check if a UTXO is frozen.

Args: outpoint: Outpoint string in txid:vout format.

Returns: True if frozen, False otherwise.

Source code in jmwallet/src/jmwallet/wallet/service.py

def is_utxo_frozen(self, outpoint: str) -> bool:
    """Check if a UTXO is frozen.

    Args:
        outpoint: Outpoint string in ``txid:vout`` format.

    Returns:
        True if frozen, False otherwise.
    """
    if self.metadata_store is None:
        return False
    return self.metadata_store.is_frozen(outpoint)

release_coinjoin_inputs(outpoints: set[tuple[str, int]]) -> None

Release CoinJoin locks held on outpoints (no-op if none).

Source code in jmwallet/src/jmwallet/wallet/service.py

def release_coinjoin_inputs(self, outpoints: set[tuple[str, int]]) -> None:
    """Release CoinJoin locks held on ``outpoints`` (no-op if none)."""
    if self.metadata_store is None or not outpoints:
        return
    refs = [f"{txid}:{vout}" for txid, vout in outpoints]
    self.metadata_store.release_outpoints(refs)

reserve_address(address: str, label: str = '') -> None

Reserve (set aside) a deposit address so it is never reissued.

Records the address in the in-memory skip-set consulted by the deposit-address pickers and, when a data_dir is configured, persists a jm:reserved record (with the optional user label) to the metadata store so the reservation survives restarts. Reserved addresses are hidden from the concise jm-wallet info view and shown with their label in the extended view.

Source code in jmwallet/src/jmwallet/wallet/service.py

def reserve_address(self, address: str, label: str = "") -> None:
    """Reserve (set aside) a deposit address so it is never reissued.

    Records the address in the in-memory skip-set consulted by the
    deposit-address pickers and, when a ``data_dir`` is configured,
    persists a ``jm:reserved`` record (with the optional user label) to the
    metadata store so the reservation survives restarts. Reserved addresses
    are hidden from the concise ``jm-wallet info`` view and shown with their
    label in the extended view.
    """
    if not address:
        return
    self.issued_receive_addresses.add(address)
    self.reserved_address_labels[address] = label or ""
    store = getattr(self, "metadata_store", None)
    if store is not None:
        try:
            store.reserve_address(address, label or "")
        except Exception as exc:  # pragma: no cover - disk failures are rare
            logger.warning(f"Failed to persist reserved address {address}: {exc}")

reserve_addresses(addresses: set[str]) -> None

Reserve addresses for an in-progress CoinJoin session.

Once addresses are shared with a taker (in !ioauth message), they must not be reused even if the CoinJoin fails. This method marks addresses as reserved so get_next_address_index() will skip past them.

Note: Addresses stay reserved until the wallet is restarted, since they may have been logged by counterparties. The CoinJoin history file provides persistent tracking across restarts.

Args: addresses: Set of addresses to reserve (typically cj_address + change_address)

Source code in jmwallet/src/jmwallet/wallet/service.py

def reserve_addresses(self, addresses: set[str]) -> None:
    """
    Reserve addresses for an in-progress CoinJoin session.

    Once addresses are shared with a taker (in !ioauth message), they must not
    be reused even if the CoinJoin fails. This method marks addresses as reserved
    so get_next_address_index() will skip past them.

    Note: Addresses stay reserved until the wallet is restarted, since they may
    have been logged by counterparties. The CoinJoin history file provides
    persistent tracking across restarts.

    Args:
        addresses: Set of addresses to reserve (typically cj_address + change_address)
    """
    self.reserved_addresses.update(addresses)
    logger.debug(f"Reserved {len(addresses)} addresses: {addresses}")

reserve_coinjoin_inputs(outpoints: set[tuple[str, int]], ttl: float = DEFAULT_COINJOIN_LOCK_TTL) -> bool

Atomically lock outpoints for an in-flight CoinJoin.

Returns True if all were locked, False on conflict (another round already holds one of them). When no metadata store is configured the call is a no-op that returns True (locking is best-effort persistence).

Source code in jmwallet/src/jmwallet/wallet/service.py

def reserve_coinjoin_inputs(
    self,
    outpoints: set[tuple[str, int]],
    ttl: float = DEFAULT_COINJOIN_LOCK_TTL,
) -> bool:
    """Atomically lock ``outpoints`` for an in-flight CoinJoin.

    Returns True if all were locked, False on conflict (another round
    already holds one of them). When no metadata store is configured the
    call is a no-op that returns True (locking is best-effort persistence).
    """
    if self.metadata_store is None or not outpoints:
        return True
    refs = [f"{txid}:{vout}" for txid, vout in outpoints]
    return self.metadata_store.try_lock_outpoints(refs, ttl=ttl)

sync() -> dict[int, list[UTXOInfo]] async

Sync wallet (alias for sync_all for backward compatibility).

Source code in jmwallet/src/jmwallet/wallet/service.py

async def sync(self) -> dict[int, list[UTXOInfo]]:
    """Sync wallet (alias for sync_all for backward compatibility)."""
    return await self.sync_all()

toggle_freeze_utxo(outpoint: str) -> bool

Toggle frozen state of a UTXO by outpoint (persisted to disk).

Args: outpoint: Outpoint string in txid:vout format.

Returns: True if now frozen, False if now unfrozen.

Raises: RuntimeError: If no metadata store is available (no data_dir).

Source code in jmwallet/src/jmwallet/wallet/service.py

def toggle_freeze_utxo(self, outpoint: str) -> bool:
    """Toggle frozen state of a UTXO by outpoint (persisted to disk).

    Args:
        outpoint: Outpoint string in ``txid:vout`` format.

    Returns:
        True if now frozen, False if now unfrozen.

    Raises:
        RuntimeError: If no metadata store is available (no data_dir).
    """
    if self.metadata_store is None:
        raise RuntimeError("Cannot toggle freeze without a data directory")
    now_frozen = self.metadata_store.toggle_freeze(outpoint)
    # Update the in-memory UTXO cache
    for utxos in self.utxo_cache.values():
        for utxo in utxos:
            if utxo.outpoint == outpoint:
                utxo.frozen = now_frozen
                break
    return now_frozen

unfreeze_utxo(outpoint: str) -> None

Unfreeze a UTXO by outpoint (persisted to disk).

Args: outpoint: Outpoint string in txid:vout format.

Raises: RuntimeError: If no metadata store is available (no data_dir).

Source code in jmwallet/src/jmwallet/wallet/service.py

def unfreeze_utxo(self, outpoint: str) -> None:
    """Unfreeze a UTXO by outpoint (persisted to disk).

    Args:
        outpoint: Outpoint string in ``txid:vout`` format.

    Raises:
        RuntimeError: If no metadata store is available (no data_dir).
    """
    if self.metadata_store is None:
        raise RuntimeError("Cannot unfreeze UTXOs without a data directory")
    self.metadata_store.unfreeze(outpoint)
    # Update the in-memory UTXO cache
    for utxos in self.utxo_cache.values():
        for utxo in utxos:
            if utxo.outpoint == outpoint:
                utxo.frozen = False
                return

unreserve_address(address: str) -> bool

Remove a reservation so the address may be reissued.

Note: an address that has real on-chain history is still never reissued (the pickers also consult addresses_with_history); this only clears the "set aside" marker and its label.

Source code in jmwallet/src/jmwallet/wallet/service.py

def unreserve_address(self, address: str) -> bool:
    """Remove a reservation so the address may be reissued.

    Note: an address that has real on-chain history is still never
    reissued (the pickers also consult ``addresses_with_history``); this
    only clears the "set aside" marker and its label.
    """
    changed = self.reserved_address_labels.pop(address, None) is not None
    self.issued_receive_addresses.discard(address)
    store = getattr(self, "metadata_store", None)
    if store is not None:
        try:
            if store.unreserve_address(address):
                changed = True
        except Exception as exc:  # pragma: no cover - disk failures are rare
            logger.warning(f"Failed to remove reserved address {address}: {exc}")
    return changed

Functions: