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 = "",
    ):
        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

        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]] = {}

        # 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)

        # 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()
        # Cache for fidelity bond locktimes (address -> locktime)
        self.fidelity_bond_locktime_cache: dict[str, int] = {}

        # 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)

    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"""
        if address not in self.address_cache:
            return None

        mixdepth, change, index = self.address_cache[address]
        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 total balance across all mixdepths.

        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:
            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, [])
        return sum(utxo.value for utxo in utxos if utxo.is_fidelity_bond and not utxo.frozen)

    # -- 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.issued_receive_addresses.add(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 registers the
        chosen address in ``issued_receive_addresses`` (matching the
        side effect of :meth:`get_new_address`). 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.issued_receive_addresses.add(address)
        return address

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

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

    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

metadata_store: UTXOMetadataStore | None = None instance-attribute

mixdepth_count = mixdepth_count instance-attribute

network = network 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

Functions

__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 = '')

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 = "",
):
    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

    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]] = {}

    # 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)

    # 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()
    # Cache for fidelity bond locktimes (address -> locktime)
    self.fidelity_bond_locktime_cache: dict[str, int] = {}

    # 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)

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: Frozen UTXOs are excluded from balance calculations.

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:
        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, [])
    return sum(utxo.value for utxo in utxos if utxo.is_fidelity_bond and not utxo.frozen)

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"""
    if address not in self.address_cache:
        return None

    mixdepth, change, index = self.address_cache[address]
    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.issued_receive_addresses.add(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 registers the chosen address in issued_receive_addresses (matching the side effect of :meth:get_new_address). 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 registers the
    chosen address in ``issued_receive_addresses`` (matching the
    side effect of :meth:`get_new_address`). 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.issued_receive_addresses.add(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_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 total balance across all mixdepths.

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 total balance across all mixdepths.

    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_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_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

Functions