crypto_watcher: dist clients for XMR/BTC/LTC/DOGE via erldistpy

Drop-in alternatives to MoneroClient + DogecoinClient (plus first-class
BTC + LTC clients) that route wallet RPC through Wallet.Service on
cammy via Erlang dist instead of hitting daemons directly.

Why both transports coexist:
  - Direct-HTTP clients keep working unchanged — no shop is forced over
  - Each shop opts in by config; once a shop's on dist, its daemons
    don't have to live on the MPS droplet
  - End goal: drain wallet daemons off prod MPS, right-size hosting

What's here:
  - ErlangDistMoneroClient mirrors MoneroClient's public surface
    (create_subaddress, get_transfers_for_subaddr, get_height,
     is_synced, get_sync_status, get_balance, sweep_subaddress,
     transfer, plus new sweep_subaddress_with_details +
     get_tx_confirmations helpers)
  - ErlangDistBitcoinClient / ErlangDistLitecoinClient /
    ErlangDistDogecoinClient share a UTXO base class, identical to
    DogecoinClient's surface (getnewaddress, getreceivedbyaddress,
    sendmany, sendtoaddress, gettransaction, listtransactions,
    is_synced, get_sync_status, plus new get_tx_confirmations)
  - refund_with_fee_split — one transaction for customer refund plus
    optional shop fee output, covers all four coins
  - refund_economically_viable — pure helper matching
    Wallet.Service.refund_economically_viable?/3 default ($0.069)

Wire shape against Wallet.Service handle_call clauses:
  {:monero, :function_name, [args...]}
  {:utxo, :coin_atom, :function_name, [args...]}
  {:refund_with_fee_split, coin, refund_addr, refund_amt, shop_addr, shop_amt, opts}

Lazy import: erldistpy is only imported when a dist client is
constructed, so MPS deploys that stay on HTTP transport don't need it
installed. Secrets discipline matches existing clients — cookie + TLS
material are paths the caller resolves; no PEM bytes enter Python.

44 unit tests covering each client's tuple shape + reply decoding +
error mapping (CallTimeout → WalletDistError, NodeError →
WalletDistError). The cross-stack wire contract is already verified
in wallet.unsandbox.com's test/functional suite (Python erldistpy →
Erlang dist → Wallet.Service → mock daemons → reply back), so MPS-side
tests focus on the adapter.

Existing 110 crypto_watcher tests untouched and still green.

Not in this commit:
  - No watcher orchestration changes (still uses direct clients)
  - No CryptoProcessor model column / Alembic migration
  - No factory wiring per-shop transport selection
That's the next discrete ticket so each piece is reviewable on its own.

make_post_sell/lib/crypto_watcher/erldist_clients.py

@@ -0,0 +1,655 @@
+"""
+Native Erlang distribution clients for Wallet.Service on wallet@cammy.
+
+Drop-in alternatives to ``MoneroClient`` and ``DogecoinClient`` (and new
+``BitcoinClient`` / ``LitecoinClient`` surfaces) that route wallet RPC
+through Wallet.Service on cammy via Erlang dist (``erldistpy``) instead
+of hitting daemons directly. The public method signatures mirror the
+existing direct-HTTP clients so the watcher loop can swap transports
+per shop without further changes.
+
+Why both transports exist:
+  - Existing direct-HTTP clients (this file's siblings) talk to
+    daemons co-located with the MPS Pyramid process. Self-contained,
+    no external dependency.
+  - These dist clients call into Wallet.Service on cammy, which owns
+    the daemon credentials. Lets us drain wallet daemons off the MPS
+    droplet to right-size hosting; each shop opts in by setting a
+    transport config.
+
+Lazy import: ``erldistpy`` is only imported when a dist client is
+constructed. MPS deploys that stick with the HTTP transport never need
+erldistpy installed.
+
+Operation Voyeur: ``cookie`` is a string the caller has already read
+from a file path. TLS material (``tls_cert``, ``tls_key``, ``tls_ca``)
+are file paths loaded by OpenSSL — no PEM bytes enter Python memory.
+"""
+
+from __future__ import annotations
+
+from decimal import Decimal
+from typing import Any, Dict, List, Optional, Tuple
+
+
+WALLET_SERVICE_NAME = "Elixir.Wallet.Service"
+
+
+def _make_node(
+    *,
+    our_name: str,
+    peer_name: str,
+    cookie: str,
+    peer_host: str = "localhost",
+    connect_timeout: float = 5.0,
+    tls_cert: Optional[str] = None,
+    tls_key: Optional[str] = None,
+    tls_ca: Optional[str] = None,
+):
+    """Open an ``erldistpy.Node`` against the wallet node.
+
+    Lazy import so ``erldistpy`` is only required when a dist client is
+    actually constructed.
+    """
+    from erldistpy import Node, make_dist_tls_context
+
+    tls_context = None
+    if tls_cert and tls_key and tls_ca:
+        tls_context = make_dist_tls_context(cert=tls_cert, key=tls_key, ca=tls_ca)
+
+    return Node(
+        our_name=our_name,
+        peer_name=peer_name,
+        cookie=cookie,
+        peer_host=peer_host,
+        connect_timeout=connect_timeout,
+        tls_context=tls_context,
+    )
+
+
+def _atom(name: str):
+    """Build an ETF atom term. Lazy import for the same reason as Node."""
+    from erldistpy import Atom
+
+    return Atom(name)
+
+
+def _to_python(value: Any) -> Any:
+    """Translate ETF-decoded terms back to JSON-shaped Python values.
+
+    Atoms decode to their name string. Binaries to utf-8 strings when
+    decodable (fall back to raw bytes). Maps/lists/tuples recurse so
+    the watcher sees the same dict shape it gets from the HTTP client.
+    """
+    from erldistpy import Atom
+
+    if isinstance(value, Atom):
+        return value.name
+    if isinstance(value, bytes):
+        try:
+            return value.decode("utf-8")
+        except UnicodeDecodeError:
+            return value
+    if isinstance(value, dict):
+        return {_to_python(k): _to_python(v) for k, v in value.items()}
+    if isinstance(value, list):
+        return [_to_python(x) for x in value]
+    if isinstance(value, tuple):
+        return tuple(_to_python(x) for x in value)
+    return value
+
+
+def _unwrap_ok(reply: Any, what: str) -> Any:
+    """Unwrap ``{:ok, value}`` or raise on ``{:error, ...}``.
+
+    Wallet.Service replies match one of these shapes:
+      - ``(Atom("ok"), value)`` → return ``_to_python(value)``
+      - ``(Atom("error"), reason)`` → raise ``WalletDistError``
+      - bare value (e.g. integer/string for direct RPC passthroughs) →
+        return as-is after ``_to_python``
+    """
+    if isinstance(reply, tuple) and len(reply) == 2:
+        head = reply[0]
+        from erldistpy import Atom
+
+        if isinstance(head, Atom):
+            if head.name == "ok":
+                return _to_python(reply[1])
+            if head.name == "error":
+                detail = _to_python(reply[1])
+                raise WalletDistError(f"{what} failed: {detail!r}")
+    # Some Wallet.Service handlers return the raw value (no ok-tuple),
+    # e.g. utxo passthroughs that return what bitcoind sent.
+    return _to_python(reply)
+
+
+class WalletDistError(RuntimeError):
+    """Wallet.Service returned an ``{:error, _}`` tuple."""
+
+
+# ---------------------------------------------------------------------------
+# Connection wrapper — holds the erldistpy.Node and exposes call(msg)
+# ---------------------------------------------------------------------------
+
+
+class _DistConn:
+    """Thin wrapper so client classes can share one Node instance.
+
+    Use ``open()`` to construct, ``close()`` when done. Each ``.call()``
+    forwards to the Wallet.Service registered process and translates
+    timeouts to a domain error.
+    """
+
+    def __init__(
+        self,
+        *,
+        our_name: str,
+        peer_name: str,
+        cookie: str,
+        peer_host: str = "localhost",
+        call_timeout: float = 15.0,
+        connect_timeout: float = 5.0,
+        tls_cert: Optional[str] = None,
+        tls_key: Optional[str] = None,
+        tls_ca: Optional[str] = None,
+    ):
+        self._call_timeout = call_timeout
+        self._node = _make_node(
+            our_name=our_name,
+            peer_name=peer_name,
+            cookie=cookie,
+            peer_host=peer_host,
+            connect_timeout=connect_timeout,
+            tls_cert=tls_cert,
+            tls_key=tls_key,
+            tls_ca=tls_ca,
+        )
+
+    def call(self, msg: Any, timeout: Optional[float] = None) -> Any:
+        from erldistpy import CallTimeout, NodeError
+
+        try:
+            return self._node.call(
+                WALLET_SERVICE_NAME, msg, timeout=timeout or self._call_timeout
+            )
+        except CallTimeout as e:
+            raise WalletDistError(f"wallet RPC timed out: {e}") from e
+        except NodeError as e:
+            raise WalletDistError(f"wallet RPC error: {e}") from e
+
+    def close(self) -> None:
+        self._node.close()
+
+    def __enter__(self) -> "_DistConn":
+        return self
+
+    def __exit__(self, *_a) -> None:
+        self.close()
+
+
+# ---------------------------------------------------------------------------
+# Monero
+# ---------------------------------------------------------------------------
+
+
+class ErlangDistMoneroClient:
+    """Drop-in replacement for ``MoneroClient`` over Erlang dist.
+
+    Mirrors ``MoneroClient``'s public surface so the crypto_watcher loop
+    can swap transports without code changes.
+    """
+
+    def __init__(
+        self,
+        *,
+        our_name: str,
+        peer_name: str,
+        cookie: str,
+        peer_host: str = "localhost",
+        call_timeout: float = 15.0,
+        connect_timeout: float = 5.0,
+        tls_cert: Optional[str] = None,
+        tls_key: Optional[str] = None,
+        tls_ca: Optional[str] = None,
+    ):
+        self._conn = _DistConn(
+            our_name=our_name,
+            peer_name=peer_name,
+            cookie=cookie,
+            peer_host=peer_host,
+            call_timeout=call_timeout,
+            connect_timeout=connect_timeout,
+            tls_cert=tls_cert,
+            tls_key=tls_key,
+            tls_ca=tls_ca,
+        )
+
+    def close(self) -> None:
+        self._conn.close()
+
+    def __enter__(self) -> "ErlangDistMoneroClient":
+        return self
+
+    def __exit__(self, *_a) -> None:
+        self.close()
+
+    # ------ Address creation
+    def create_subaddress(
+        self, account_index: int = 0, label: Optional[str] = None
+    ) -> Tuple[str, int]:
+        msg = (
+            _atom("monero"),
+            _atom("create_subaddress"),
+            [account_index, label.encode("utf-8") if label else None],
+        )
+        result = _unwrap_ok(self._conn.call(msg), "create_subaddress")
+        # Wallet.Service returns {:ok, {address, index}} → (str, int) tuple
+        if isinstance(result, tuple) and len(result) == 2:
+            address, index = result
+            return (
+                address if isinstance(address, str) else address.decode("utf-8"),
+                int(index),
+            )
+        raise WalletDistError(f"create_subaddress: unexpected reply shape {result!r}")
+
+    # ------ Payment polling
+    def get_transfers_for_subaddr(
+        self, account_index: int, subaddr_indices: List[int]
+    ) -> Dict[str, Any]:
+        msg = (
+            _atom("monero"),
+            _atom("get_transfers_for_subaddr"),
+            [account_index, list(subaddr_indices)],
+        )
+        return _unwrap_ok(self._conn.call(msg), "get_transfers_for_subaddr") or {}
+
+    # ------ Balance
+    def get_balance(
+        self, account_index: int = 0, subaddress_indices: Optional[List[int]] = None
+    ) -> Dict[str, Any]:
+        msg = (
+            _atom("monero"),
+            _atom("get_balance"),
+            [account_index, list(subaddress_indices or [])],
+        )
+        return _unwrap_ok(self._conn.call(msg), "get_balance") or {}
+
+    # ------ Height + sync
+    def get_height(self) -> int:
+        msg = (_atom("monero"), _atom("get_height"), [])
+        result = _unwrap_ok(self._conn.call(msg), "get_height")
+        return int(result) if result is not None else 0
+
+    def is_synced(self) -> bool:
+        try:
+            return self.get_height() > 0
+        except Exception:
+            return False
+
+    def get_sync_status(self) -> dict:
+        try:
+            h = self.get_height()
+            synced = h > 0
+            return {
+                "wallet_height": h,
+                "synced": synced,
+                "sync_percentage": 100.0 if synced else 0.0,
+                "remote_node": True,
+                "ready": synced,
+            }
+        except Exception as e:
+            return {
+                "wallet_height": 0,
+                "synced": False,
+                "sync_percentage": 0.0,
+                "remote_node": True,
+                "ready": False,
+                "error": str(e),
+            }
+
+    def refresh(self) -> Any:
+        msg = (_atom("monero"), _atom("refresh"), [])
+        return _unwrap_ok(self._conn.call(msg, timeout=60.0), "refresh")
+
+    # ------ Sweep
+    def sweep_subaddress(
+        self,
+        account_index: int,
+        subaddress_index: int,
+        destination: str,
+        priority: int = 0,
+    ) -> str:
+        msg = (
+            _atom("monero"),
+            _atom("sweep_subaddress"),
+            [account_index, subaddress_index, destination.encode("utf-8"), priority],
+        )
+        result = _unwrap_ok(self._conn.call(msg, timeout=30.0), "sweep_subaddress")
+        return result if isinstance(result, str) else str(result)
+
+    def sweep_subaddress_with_details(
+        self,
+        account_index: int,
+        subaddress_index: int,
+        destination: str,
+        priority: int = 0,
+    ) -> Dict[str, Any]:
+        """Returns ``{"tx_hash": str, "amount": int}`` for fee accounting.
+
+        Use this instead of ``sweep_subaddress`` whenever you need the
+        actual swept amount (e.g. restocking-fee math against the
+        on-chain amount after network fees).
+        """
+        msg = (
+            _atom("monero"),
+            _atom("sweep_subaddress_with_details"),
+            [account_index, subaddress_index, destination.encode("utf-8"), priority],
+        )
+        return _unwrap_ok(
+            self._conn.call(msg, timeout=30.0), "sweep_subaddress_with_details"
+        ) or {}
+
+    # ------ Transfer (refunds)
+    def transfer(
+        self,
+        destinations: List[Dict[str, Any]],
+        account_index: int = 0,
+        priority: int = 0,
+        get_tx_key: bool = True,
+    ) -> Dict[str, Any]:
+        """``destinations``: ``[{"amount": int_atomic, "address": str}, ...]``."""
+        encoded = [
+            {b"amount": int(d["amount"]), b"address": d["address"].encode("utf-8")}
+            for d in destinations
+        ]
+        msg = (
+            _atom("monero"),
+            _atom("transfer"),
+            [
+                encoded,
+                account_index,
+                [(_atom("priority"), priority), (_atom("get_tx_key"), get_tx_key)],
+            ],
+        )
+        return _unwrap_ok(self._conn.call(msg, timeout=30.0), "transfer") or {}
+
+    # ------ Confirmation tracking (new helper)
+    def get_tx_confirmations(self, tx_hash: str, account_index: int = 0) -> int:
+        msg = (
+            _atom("monero"),
+            _atom("get_tx_confirmations"),
+            [tx_hash.encode("utf-8"), account_index],
+        )
+        result = _unwrap_ok(self._conn.call(msg), "get_tx_confirmations")
+        return int(result) if result is not None else 0
+
+
+# ---------------------------------------------------------------------------
+# UTXO coins — shared base, one subclass per coin
+# ---------------------------------------------------------------------------
+
+
+class _ErlangDistUtxoClient:
+    """Shared base for Bitcoin / Litecoin / Dogecoin dist clients.
+
+    All three speak the same bitcoind-style RPC contract; only the coin
+    atom (``:btc`` / ``:ltc`` / ``:doge``) differs. Subclasses set
+    ``COIN_ATOM`` to identify themselves to Wallet.Service.
+    """
+
+    COIN_ATOM: str = ""  # overridden by subclass
+
+    def __init__(
+        self,
+        *,
+        our_name: str,
+        peer_name: str,
+        cookie: str,
+        peer_host: str = "localhost",
+        call_timeout: float = 15.0,
+        connect_timeout: float = 5.0,
+        tls_cert: Optional[str] = None,
+        tls_key: Optional[str] = None,
+        tls_ca: Optional[str] = None,
+    ):
+        if not self.COIN_ATOM:
+            raise TypeError(
+                f"{type(self).__name__}: COIN_ATOM must be set on subclass"
+            )
+        self._conn = _DistConn(
+            our_name=our_name,
+            peer_name=peer_name,
+            cookie=cookie,
+            peer_host=peer_host,
+            call_timeout=call_timeout,
+            connect_timeout=connect_timeout,
+            tls_cert=tls_cert,
+            tls_key=tls_key,
+            tls_ca=tls_ca,
+        )
+
+    def close(self) -> None:
+        self._conn.close()
+
+    def __enter__(self) -> "_ErlangDistUtxoClient":
+        return self
+
+    def __exit__(self, *_a) -> None:
+        self.close()
+
+    def _utxo_call(self, function: str, args: List[Any], timeout: Optional[float] = None) -> Any:
+        msg = (
+            _atom("utxo"),
+            _atom(self.COIN_ATOM),
+            _atom(function),
+            args,
+        )
+        return _unwrap_ok(self._conn.call(msg, timeout=timeout), function)
+
+    # ------ Wallet management
+    def getnewaddress(self, label: str = "") -> str:
+        result = self._utxo_call("getnewaddress", [label.encode("utf-8")])
+        return result if isinstance(result, str) else str(result)
+
+    def getaddressesbylabel(self, label: str) -> Dict[str, Any]:
+        return self._utxo_call("getaddressesbylabel", [label.encode("utf-8")]) or {}
+
+    def validateaddress(self, address: str) -> Dict[str, Any]:
+        return self._utxo_call("validateaddress", [address.encode("utf-8")]) or {}
+
+    # ------ Balances + history
+    def getbalance(self) -> float:
+        result = self._utxo_call("getbalance", [])
+        return float(result) if result is not None else 0.0
+
+    def getreceivedbyaddress(self, address: str, minconf: int = 0) -> float:
+        result = self._utxo_call(
+            "getreceivedbyaddress", [address.encode("utf-8"), minconf]
+        )
+        return float(result) if result is not None else 0.0
+
+    def listtransactions(
+        self, label: str = "*", count: int = 10, skip: int = 0
+    ) -> List[Dict[str, Any]]:
+        return self._utxo_call(
+            "listtransactions", [label.encode("utf-8"), count, skip]
+        ) or []
+
+    def gettransaction(self, txid: str) -> Dict[str, Any]:
+        return self._utxo_call("gettransaction", [txid.encode("utf-8")]) or {}
+
+    # ------ Sending
+    def sendtoaddress(self, address: str, amount: float, comment: str = "") -> str:
+        opts = []
+        if comment:
+            opts.append((_atom("comment"), comment.encode("utf-8")))
+        result = self._utxo_call(
+            "sendtoaddress",
+            [address.encode("utf-8"), float(amount), opts],
+            timeout=30.0,
+        )
+        return result if isinstance(result, str) else str(result)
+
+    def sendmany(
+        self,
+        from_label: str,
+        addresses_amounts: Dict[str, float],
+        minconf: int = 1,
+        comment: str = "",
+    ) -> str:
+        outputs = {
+            addr.encode("utf-8"): float(amt) for addr, amt in addresses_amounts.items()
+        }
+        opts = [(_atom("minconf"), minconf), (_atom("from_account"), from_label.encode("utf-8"))]
+        if comment:
+            opts.append((_atom("comment"), comment.encode("utf-8")))
+        result = self._utxo_call("sendmany", [outputs, opts], timeout=30.0)
+        return result if isinstance(result, str) else str(result)
+
+    # ------ Blockchain status
+    def getblockcount(self) -> int:
+        # Some Wallet.Service handlers may not expose getblockcount —
+        # fall back to getblockchaininfo if it errors.
+        try:
+            result = self._utxo_call("getblockcount", [])
+            return int(result) if result is not None else 0
+        except WalletDistError:
+            info = self.getblockchaininfo()
+            return int(info.get("blocks", 0))
+
+    def getblockchaininfo(self) -> Dict[str, Any]:
+        return self._utxo_call("getblockchaininfo", []) or {}
+
+    def is_synced(self) -> bool:
+        try:
+            info = self.getblockchaininfo()
+            blocks = int(info.get("blocks") or 0)
+            headers = int(info.get("headers") or 0)
+            return blocks > 0 and blocks >= headers - 1
+        except Exception:
+            return False
+
+    def get_sync_status(self) -> dict:
+        try:
+            info = self.getblockchaininfo()
+            blocks = int(info.get("blocks") or 0)
+            headers = int(info.get("headers") or 0)
+            synced = blocks > 0 and blocks >= headers - 1
+            return {
+                "blocks": blocks,
+                "headers": headers,
+                "synced": synced,
+                "sync_percentage": 100.0 * blocks / headers if headers else 0.0,
+                "ready": synced,
+            }
+        except Exception as e:
+            return {
+                "blocks": 0,
+                "headers": 0,
+                "synced": False,
+                "sync_percentage": 0.0,
+                "ready": False,
+                "error": str(e),
+            }
+
+    # ------ Confirmation tracking (new helper)
+    def get_tx_confirmations(self, txid: str) -> int:
+        result = self._utxo_call("get_tx_confirmations", [txid.encode("utf-8")])
+        return int(result) if result is not None else 0
+
+
+class ErlangDistBitcoinClient(_ErlangDistUtxoClient):
+    COIN_ATOM = "btc"
+
+
+class ErlangDistLitecoinClient(_ErlangDistUtxoClient):
+    COIN_ATOM = "ltc"
+
+
+class ErlangDistDogecoinClient(_ErlangDistUtxoClient):
+    COIN_ATOM = "doge"
+
+
+# ---------------------------------------------------------------------------
+# Cross-coin helpers (operate on coin_type parameter, not tied to one client)
+# ---------------------------------------------------------------------------
+
+
+def refund_with_fee_split(
+    conn: _DistConn,
+    coin_type: str,
+    refund_address: str,
+    refund_amount_atomic: int,
+    shop_address: Optional[str] = None,
+    shop_amount_atomic: int = 0,
+    opts: Optional[List[Tuple[Any, Any]]] = None,
+) -> Dict[str, Any]:
+    """Single-transaction customer refund + optional shop fee output.
+
+    ``coin_type``: one of ``"xmr"``, ``"btc"``, ``"ltc"``, ``"doge"``.
+
+    Atomic amounts (piconero / satoshi / litoshi / koinu). Pass
+    ``shop_amount_atomic=0`` for a refund without a fee split.
+
+    Returns ``{"tx_hash": str}`` on success or raises ``WalletDistError``.
+    """
+    msg = (
+        _atom("refund_with_fee_split"),
+        _atom(coin_type),
+        refund_address.encode("utf-8"),
+        int(refund_amount_atomic),
+        shop_address.encode("utf-8") if shop_address else None,
+        int(shop_amount_atomic),
+        opts or [],
+    )
+    return _unwrap_ok(conn.call(msg, timeout=30.0), "refund_with_fee_split") or {}
+
+
+def refund_economically_viable(
+    refund_amount_coin: Any,
+    usd_per_coin: Optional[Any],
+    min_usd: Any = Decimal("0.069"),
+) -> bool:
+    """Pure helper — same math as Wallet.Service.refund_economically_viable?/3.
+
+    Doesn't touch the network. Lives here so callers don't have to round-
+    trip a trivial decision through dist. Wallet.Service exposes the
+    same predicate so the Elixir side has a single source of truth, but
+    Python callers can short-circuit.
+
+    Default ``min_usd`` matches MPS's ``MINIMUM_VIABLE_REFUND_USD``.
+    """
+    if usd_per_coin is None:
+        return True
+    refund = refund_amount_coin if isinstance(refund_amount_coin, Decimal) else Decimal(str(refund_amount_coin))
+    rate = usd_per_coin if isinstance(usd_per_coin, Decimal) else Decimal(str(usd_per_coin))
+    threshold = min_usd if isinstance(min_usd, Decimal) else Decimal(str(min_usd))
+    return (refund * rate) >= threshold
+
+
+def open_wallet_conn(
+    *,
+    our_name: str,
+    peer_name: str,
+    cookie: str,
+    peer_host: str = "localhost",
+    call_timeout: float = 15.0,
+    connect_timeout: float = 5.0,
+    tls_cert: Optional[str] = None,
+    tls_key: Optional[str] = None,
+    tls_ca: Optional[str] = None,
+) -> _DistConn:
+    """Open a connection for callers that want to use ``refund_with_fee_split``
+    directly (without holding a per-coin client). Caller closes it with
+    ``.close()`` or uses it as a context manager."""
+    return _DistConn(
+        our_name=our_name,
+        peer_name=peer_name,
+        cookie=cookie,
+        peer_host=peer_host,
+        call_timeout=call_timeout,
+        connect_timeout=connect_timeout,
+        tls_cert=tls_cert,
+        tls_key=tls_key,
+        tls_ca=tls_ca,
+    )

make_post_sell/tests/test_erldist_clients.py

@@ -0,0 +1,447 @@
+"""Tests for ``erldist_clients`` — dist-transport wallet clients.
+
+The Wallet.Service cross-stack tests in ~/git/wallet.unsandbox.com
+already prove the wire contract works end-to-end against a real
+Elixir node + mock daemons. These tests assert the MPS-side adapter:
+
+  - Each client builds the correct control tuple for Wallet.Service's
+    handle_call clauses
+  - ``{:ok, value}`` replies decode to Python dict/str/int matching the
+    HTTP client's contract (atoms → str, binaries → str, recursive)
+  - ``{:error, _}`` replies raise WalletDistError
+  - CallTimeout / NodeError from erldistpy surface as WalletDistError
+
+We mock ``erldistpy.Node`` rather than running a real wallet node — that
+end-to-end path is covered by the wallet.unsandbox.com cross-stack suite.
+"""
+
+from __future__ import annotations
+
+from decimal import Decimal
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+# Lazy imports inside the module under test; we still need the symbols
+# for tests directly. Importing them at module load is OK because
+# erldistpy *is* installed in the test venv (or skipped if not).
+try:
+    from erldistpy import Atom, CallTimeout, NodeError
+except ImportError:  # pragma: no cover
+    pytest.skip("erldistpy not installed", allow_module_level=True)
+
+from make_post_sell.lib.crypto_watcher.erldist_clients import (
+    ErlangDistBitcoinClient,
+    ErlangDistDogecoinClient,
+    ErlangDistLitecoinClient,
+    ErlangDistMoneroClient,
+    WalletDistError,
+    refund_economically_viable,
+    refund_with_fee_split,
+)
+
+
+# --------------------------------------------------------------------------
+# Test infrastructure — replace _DistConn.call so we don't open a real Node
+# --------------------------------------------------------------------------
+
+
+class FakeConn:
+    """Minimal stand-in for ``_DistConn`` that records calls and returns
+    canned replies."""
+
+    def __init__(self):
+        self.calls = []
+        self._replies = []
+
+    def will_reply(self, *replies):
+        """Queue one or more replies; consumed FIFO on each ``call``."""
+        self._replies.extend(replies)
+
+    def call(self, msg, timeout=None):
+        self.calls.append({"msg": msg, "timeout": timeout})
+        if not self._replies:
+            raise AssertionError(f"unexpected call (no reply queued): {msg!r}")
+        reply = self._replies.pop(0)
+        if isinstance(reply, Exception):
+            raise reply
+        return reply
+
+    def close(self):
+        pass
+
+
+@pytest.fixture
+def fake_conn():
+    return FakeConn()
+
+
+def _attach_fake_conn(client, fake_conn):
+    """Replace the freshly-constructed ``_DistConn`` on a client with our fake."""
+    client._conn = fake_conn
+
+
+def _make_client(cls, **overrides):
+    """Construct a dist client without opening a real Node connection."""
+    # Patch erldistpy.Node so __init__ doesn't try to dial cammy.
+    with patch("erldistpy.Node") as ctor:
+        ctor.return_value = MagicMock()
+        defaults = dict(
+            our_name="mps@local",
+            peer_name="wallet",
+            cookie="X",
+        )
+        defaults.update(overrides)
+        return cls(**defaults)
+
+
+# --------------------------------------------------------------------------
+# ErlangDistMoneroClient
+# --------------------------------------------------------------------------
+
+
+class TestMoneroCreateSubaddress:
+    def test_returns_address_and_index_tuple(self, fake_conn):
+        fake_conn.will_reply((Atom("ok"), (b"4xmraddress", 7)))
+        client = _make_client(ErlangDistMoneroClient)
+        _attach_fake_conn(client, fake_conn)
+
+        addr, idx = client.create_subaddress(account_index=0, label="checkout-42")
+
+        assert addr == "4xmraddress"
+        assert idx == 7
+        sent_msg = fake_conn.calls[0]["msg"]
+        assert sent_msg[0] == Atom("monero")
+        assert sent_msg[1] == Atom("create_subaddress")
+        assert sent_msg[2] == [0, b"checkout-42"]
+
+    def test_none_label_passed_through(self, fake_conn):
+        fake_conn.will_reply((Atom("ok"), (b"4xmraddress", 1)))
+        client = _make_client(ErlangDistMoneroClient)
+        _attach_fake_conn(client, fake_conn)
+
+        client.create_subaddress(account_index=0)
+
+        sent_msg = fake_conn.calls[0]["msg"]
+        # None label preserved so Elixir side picks its default
+        assert sent_msg[2] == [0, None]
+
+    def test_error_reply_raises(self, fake_conn):
+        fake_conn.will_reply((Atom("error"), Atom("wallet_not_ready")))
+        client = _make_client(ErlangDistMoneroClient)
+        _attach_fake_conn(client, fake_conn)
+
+        with pytest.raises(WalletDistError, match="create_subaddress failed"):
+            client.create_subaddress()
+
+
+class TestMoneroGetTransfers:
+    def test_returns_decoded_map(self, fake_conn):
+        fake_conn.will_reply(
+            (
+                Atom("ok"),
+                {
+                    b"in": [
+                        {b"amount": 1_000_000_000, b"txid": b"tx1", b"confirmations": 10}
+                    ],
+                    b"pool": [],
+                },
+            )
+        )
+        client = _make_client(ErlangDistMoneroClient)
+        _attach_fake_conn(client, fake_conn)
+
+        result = client.get_transfers_for_subaddr(0, [3, 4, 5])
+
+        assert result["in"][0]["txid"] == "tx1"
+        assert result["in"][0]["amount"] == 1_000_000_000
+        sent_msg = fake_conn.calls[0]["msg"]
+        assert sent_msg[2] == [0, [3, 4, 5]]
+
+
+class TestMoneroGetHeight:
+    def test_returns_int(self, fake_conn):
+        fake_conn.will_reply((Atom("ok"), 3_456_789))
+        client = _make_client(ErlangDistMoneroClient)
+        _attach_fake_conn(client, fake_conn)
+
+        assert client.get_height() == 3_456_789
+
+
+class TestMoneroIsSynced:
+    def test_synced_when_height_positive(self, fake_conn):
+        fake_conn.will_reply((Atom("ok"), 100))
+        client = _make_client(ErlangDistMoneroClient)
+        _attach_fake_conn(client, fake_conn)
+        assert client.is_synced() is True
+
+    def test_not_synced_on_error(self, fake_conn):
+        fake_conn.will_reply((Atom("error"), Atom("wallet_disconnected")))
+        client = _make_client(ErlangDistMoneroClient)
+        _attach_fake_conn(client, fake_conn)
+        # is_synced swallows errors and returns False (matches existing
+        # MoneroClient semantics)
+        assert client.is_synced() is False
+
+
+class TestMoneroSweepSubaddress:
+    def test_returns_tx_hash_string(self, fake_conn):
+        fake_conn.will_reply((Atom("ok"), b"sweep-tx-abc"))
+        client = _make_client(ErlangDistMoneroClient)
+        _attach_fake_conn(client, fake_conn)
+
+        tx = client.sweep_subaddress(0, 5, "cold-xmr", priority=1)
+
+        assert tx == "sweep-tx-abc"
+        assert fake_conn.calls[0]["timeout"] == 30.0
+
+
+class TestMoneroSweepSubaddressWithDetails:
+    def test_returns_amount_and_tx_hash(self, fake_conn):
+        fake_conn.will_reply(
+            (Atom("ok"), {Atom("tx_hash"): b"tx-7", Atom("amount"): 12_345_678})
+        )
+        client = _make_client(ErlangDistMoneroClient)
+        _attach_fake_conn(client, fake_conn)
+
+        result = client.sweep_subaddress_with_details(0, 5, "cold-xmr", priority=1)
+
+        assert result == {"tx_hash": "tx-7", "amount": 12_345_678}
+
+    def test_no_balance_error_surfaces(self, fake_conn):
+        fake_conn.will_reply((Atom("error"), Atom("no_unlocked_balance")))
+        client = _make_client(ErlangDistMoneroClient)
+        _attach_fake_conn(client, fake_conn)
+
+        with pytest.raises(WalletDistError):
+            client.sweep_subaddress_with_details(0, 5, "cold-xmr")
+
+
+class TestMoneroTransfer:
+    def test_encodes_destinations_as_binary_keyed_maps(self, fake_conn):
+        fake_conn.will_reply(
+            (Atom("ok"), {b"tx_hash": b"refund-tx", b"tx_key": b"key1"})
+        )
+        client = _make_client(ErlangDistMoneroClient)
+        _attach_fake_conn(client, fake_conn)
+
+        client.transfer(
+            destinations=[{"amount": 900, "address": "cust"}, {"amount": 100, "address": "shop"}],
+            account_index=0,
+        )
+
+        sent_destinations = fake_conn.calls[0]["msg"][2][0]
+        assert sent_destinations == [
+            {b"amount": 900, b"address": b"cust"},
+            {b"amount": 100, b"address": b"shop"},
+        ]
+
+
+class TestMoneroGetTxConfirmations:
+    def test_returns_count(self, fake_conn):
+        fake_conn.will_reply((Atom("ok"), 13))
+        client = _make_client(ErlangDistMoneroClient)
+        _attach_fake_conn(client, fake_conn)
+        assert client.get_tx_confirmations("hash1", account_index=2) == 13
+
+        sent_msg = fake_conn.calls[0]["msg"]
+        assert sent_msg[2] == [b"hash1", 2]
+
+
+# --------------------------------------------------------------------------
+# UTXO clients (Bitcoin / Litecoin / Dogecoin)
+# --------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    "client_cls,coin_atom",
+    [
+        (ErlangDistBitcoinClient, "btc"),
+        (ErlangDistLitecoinClient, "ltc"),
+        (ErlangDistDogecoinClient, "doge"),
+    ],
+)
+class TestUtxoClients:
+    """Each UTXO subclass shares behavior; parametrize once."""
+
+    def test_getnewaddress(self, client_cls, coin_atom, fake_conn):
+        fake_conn.will_reply((Atom("ok"), b"new-addr"))
+        client = _make_client(client_cls)
+        _attach_fake_conn(client, fake_conn)
+
+        addr = client.getnewaddress("shop-1:order-99")
+
+        assert addr == "new-addr"
+        msg = fake_conn.calls[0]["msg"]
+        assert msg[0] == Atom("utxo")
+        assert msg[1] == Atom(coin_atom)
+        assert msg[2] == Atom("getnewaddress")
+        assert msg[3] == [b"shop-1:order-99"]
+
+    def test_getreceivedbyaddress(self, client_cls, coin_atom, fake_conn):
+        fake_conn.will_reply((Atom("ok"), 1.234))
+        client = _make_client(client_cls)
+        _attach_fake_conn(client, fake_conn)
+
+        amt = client.getreceivedbyaddress("some-addr", minconf=3)
+
+        assert amt == 1.234
+        msg = fake_conn.calls[0]["msg"]
+        assert msg[3] == [b"some-addr", 3]
+
+    def test_gettransaction_decodes_map(self, client_cls, coin_atom, fake_conn):
+        fake_conn.will_reply(
+            (Atom("ok"), {b"confirmations": 6, b"amount": 0.5, b"txid": b"some-txid"})
+        )
+        client = _make_client(client_cls)
+        _attach_fake_conn(client, fake_conn)
+
+        tx = client.gettransaction("some-txid")
+
+        assert tx == {"confirmations": 6, "amount": 0.5, "txid": "some-txid"}
+
+    def test_get_tx_confirmations(self, client_cls, coin_atom, fake_conn):
+        fake_conn.will_reply((Atom("ok"), 7))
+        client = _make_client(client_cls)
+        _attach_fake_conn(client, fake_conn)
+
+        assert client.get_tx_confirmations("txid") == 7
+
+    def test_sendmany(self, client_cls, coin_atom, fake_conn):
+        fake_conn.will_reply((Atom("ok"), b"tx-broadcast"))
+        client = _make_client(client_cls)
+        _attach_fake_conn(client, fake_conn)
+
+        tx = client.sendmany(
+            from_label="",
+            addresses_amounts={"addr1": 1.5, "addr2": 2.5},
+            minconf=2,
+        )
+
+        assert tx == "tx-broadcast"
+        msg = fake_conn.calls[0]["msg"]
+        # outputs map encoded as binary keys + float values
+        outputs = msg[3][0]
+        assert outputs[b"addr1"] == 1.5
+        assert outputs[b"addr2"] == 2.5
+
+    def test_is_synced_true_when_blocks_caught_up(
+        self, client_cls, coin_atom, fake_conn
+    ):
+        fake_conn.will_reply(
+            (Atom("ok"), {b"blocks": 1_000_000, b"headers": 1_000_001})
+        )
+        client = _make_client(client_cls)
+        _attach_fake_conn(client, fake_conn)
+
+        assert client.is_synced() is True
+
+    def test_is_synced_false_when_error(self, client_cls, coin_atom, fake_conn):
+        fake_conn.will_reply((Atom("error"), Atom("rpc_down")))
+        client = _make_client(client_cls)
+        _attach_fake_conn(client, fake_conn)
+
+        assert client.is_synced() is False
+
+
+# --------------------------------------------------------------------------
+# Cross-coin helpers
+# --------------------------------------------------------------------------
+
+
+class TestRefundWithFeeSplit:
+    def test_xmr_single_refund(self, fake_conn):
+        fake_conn.will_reply((Atom("ok"), {Atom("tx_hash"): b"refund-tx-1"}))
+        result = refund_with_fee_split(
+            fake_conn, "xmr", "cust-addr", 1_000_000_000_000
+        )
+        assert result == {"tx_hash": "refund-tx-1"}
+
+        msg = fake_conn.calls[0]["msg"]
+        assert msg[0] == Atom("refund_with_fee_split")
+        assert msg[1] == Atom("xmr")
+        assert msg[2] == b"cust-addr"
+        assert msg[3] == 1_000_000_000_000
+        assert msg[4] is None  # no shop address
+        assert msg[5] == 0  # no shop amount
+
+    def test_doge_split_refund(self, fake_conn):
+        fake_conn.will_reply((Atom("ok"), {Atom("tx_hash"): b"split-tx"}))
+        result = refund_with_fee_split(
+            fake_conn,
+            "doge",
+            "cust-doge",
+            9_000_000_000,
+            shop_address="shop-doge",
+            shop_amount_atomic=1_000_000_000,
+        )
+        assert result == {"tx_hash": "split-tx"}
+
+        msg = fake_conn.calls[0]["msg"]
+        assert msg[4] == b"shop-doge"
+        assert msg[5] == 1_000_000_000
+
+    def test_unsupported_coin_propagates_error(self, fake_conn):
+        fake_conn.will_reply(
+            (Atom("error"), (Atom("unsupported_coin"), Atom("zec")))
+        )
+        with pytest.raises(WalletDistError):
+            refund_with_fee_split(fake_conn, "zec", "addr", 100)
+
+
+class TestRefundEconomicallyViable:
+    def test_nil_rate_always_viable(self):
+        assert refund_economically_viable(Decimal("10"), None)
+
+    def test_dust_refund_not_viable(self):
+        # 0.0001 XMR at $150 = $0.015 < $0.069
+        assert not refund_economically_viable(Decimal("0.0001"), Decimal("150"))
+
+    def test_meaningful_refund_viable(self):
+        # 1 DOGE at $0.10 = $0.10 > $0.069
+        assert refund_economically_viable(Decimal("1"), Decimal("0.10"))
+
+    def test_at_threshold_is_viable(self):
+        # exactly $0.069 -> >= comparison passes
+        assert refund_economically_viable(Decimal("1"), Decimal("0.069"))
+
+    def test_custom_min(self):
+        assert not refund_economically_viable(
+            Decimal("0.5"), Decimal("0.10"), min_usd=Decimal("0.10")
+        )
+        assert refund_economically_viable(
+            Decimal("0.5"), Decimal("0.10"), min_usd=Decimal("0.01")
+        )
+
+    def test_accepts_non_decimal_numerics(self):
+        assert refund_economically_viable(100, 1)
+        assert refund_economically_viable("10", "0.5")
+        assert not refund_economically_viable(0.001, 0.1)
+
+
+# --------------------------------------------------------------------------
+# Error mapping — CallTimeout / NodeError surface as WalletDistError
+# --------------------------------------------------------------------------
+
+
+class TestErrorMapping:
+    def test_call_timeout_becomes_wallet_dist_error(self, fake_conn):
+        fake_conn.will_reply(CallTimeout("dist timeout"))
+        client = _make_client(ErlangDistMoneroClient)
+        # We need the real _DistConn to translate exceptions — let it
+        # use the fake's call. Replace just the underlying _node.call.
+        real_conn = client._conn
+        real_conn._node = MagicMock()
+        real_conn._node.call.side_effect = CallTimeout("dist timeout")
+
+        with pytest.raises(WalletDistError, match="timed out"):
+            client.get_height()
+
+    def test_node_error_becomes_wallet_dist_error(self):
+        client = _make_client(ErlangDistMoneroClient)
+        real_conn = client._conn
+        real_conn._node = MagicMock()
+        real_conn._node.call.side_effect = NodeError("peer down")
+
+        with pytest.raises(WalletDistError, match="wallet RPC error"):
+            client.get_height()