taker.taker

taker.taker

Main Taker class for CoinJoin execution.

Orchestrates the complete CoinJoin protocol: 1. Fetch orderbook from directory nodes 2. Select makers and generate PoDLE commitment 3. Send !fill requests and receive !pubkey responses 4. Send !auth with PoDLE proof and receive !ioauth (maker UTXOs) 5. Build unsigned transaction and send !tx 6. Collect !sig responses and broadcast

Reference: Original joinmarket-clientserver/src/jmclient/taker.py

Attributes

__all__ = ['MultiDirectoryClient', 'TakerState', 'MakerSession', 'PhaseResult', 'Taker', 'warn_if_destination_script_mismatch'] module-attribute

Classes

MakerSession

Session data for a single maker.

Source code in taker/src/taker/models.py

@dataclass(config=ConfigDict(arbitrary_types_allowed=True))
class MakerSession:
    """Session data for a single maker."""

    nick: str
    offer: Offer
    utxos: list[dict[str, Any]] = Field(default_factory=list)
    cj_address: str = ""
    change_address: str = ""
    pubkey: str = ""  # Maker's NaCl public key (hex)
    auth_pubkey: str = ""  # Maker's EC auth public key from !ioauth (hex)
    crypto: CryptoSession | None = None  # Encryption session with this maker
    signature: dict[str, Any] | None = None
    responded_fill: bool = False
    responded_auth: bool = False
    responded_sig: bool = False
    supports_neutrino_compat: bool = False  # Supports extended UTXO metadata for Neutrino
    # Communication channel used for this session (must be consistent throughout)
    # "direct" = peer-to-peer onion connection
    # "directory:<host>:<port>" = relayed through specific directory
    comm_channel: str = ""

Attributes

auth_pubkey: str = '' class-attribute instance-attribute

change_address: str = '' class-attribute instance-attribute

cj_address: str = '' class-attribute instance-attribute

comm_channel: str = '' class-attribute instance-attribute

crypto: CryptoSession | None = None class-attribute instance-attribute

nick: str instance-attribute

offer: Offer instance-attribute

pubkey: str = '' class-attribute instance-attribute

responded_auth: bool = False class-attribute instance-attribute

responded_fill: bool = False class-attribute instance-attribute

responded_sig: bool = False class-attribute instance-attribute

signature: dict[str, Any] | None = None class-attribute instance-attribute

supports_neutrino_compat: bool = False class-attribute instance-attribute

utxos: list[dict[str, Any]] = Field(default_factory=list) class-attribute instance-attribute

MultiDirectoryClient

Bases: DirectoryClientPool

Wrapper for managing multiple DirectoryClient connections.

Provides a unified interface for connecting to multiple directory servers and aggregating orderbook data. Implements multi-directory aware nick tracking - a nick is only considered "gone" when ALL directories report it as disconnected.

Direct Peer Connections: When enabled (prefer_direct_connections=True), the client will establish direct Tor connections to makers when possible, bypassing directory servers for private messages. This improves privacy by preventing directories from observing who is communicating with whom.

Connection flow: 1. First message to a maker goes via directory relay 2. Opportunistically starts direct connection in background 3. Subsequent messages prefer direct connection if available 4. Falls back to directory relay if direct connection fails

This prevents premature maker removal when: - A maker temporarily disconnects from one directory but remains on others - Directory connections are flaky or experiencing network issues - There's a race condition between directory updates

Reference: JoinMarket onionmc.py lines 1078-1103

Source code in taker/src/taker/multi_directory.py

class MultiDirectoryClient(DirectoryClientPool):
    """
    Wrapper for managing multiple DirectoryClient connections.

    Provides a unified interface for connecting to multiple directory servers
    and aggregating orderbook data. Implements multi-directory aware nick
    tracking - a nick is only considered "gone" when ALL directories report
    it as disconnected.

    Direct Peer Connections:
    When enabled (prefer_direct_connections=True), the client will establish
    direct Tor connections to makers when possible, bypassing directory servers
    for private messages. This improves privacy by preventing directories from
    observing who is communicating with whom.

    Connection flow:
    1. First message to a maker goes via directory relay
    2. Opportunistically starts direct connection in background
    3. Subsequent messages prefer direct connection if available
    4. Falls back to directory relay if direct connection fails

    This prevents premature maker removal when:
    - A maker temporarily disconnects from one directory but remains on others
    - Directory connections are flaky or experiencing network issues
    - There's a race condition between directory updates

    Reference: JoinMarket onionmc.py lines 1078-1103
    """

    def __init__(
        self,
        directory_servers: list[str],
        network: str,
        nick_identity: NickIdentity,
        socks_host: str = "127.0.0.1",
        socks_port: int = 9050,
        connection_timeout: float = 120.0,
        neutrino_compat: bool = False,
        on_nick_leave: Any | None = None,
        prefer_direct_connections: bool = True,
        our_location: str = "NOT-SERVING-ONION",
        stream_isolation: bool = False,
        nick_auth_mode: NickAuthMode = NickAuthMode.PREFER_VERIFIED,
        nick_auth_directory_ids: dict[str, str] | None = None,
    ):
        # Connection / SOCKS / credential setup is delegated to the
        # DirectoryClientPool base; it handles directory_servers, network,
        # nick_identity, SOCKS params, connection_timeout, stream_isolation,
        # the clients dict, and _dir_creds / _peer_creds.
        super().__init__(
            directory_servers=directory_servers,
            network=network,
            nick_identity=nick_identity,
            socks_host=socks_host,
            socks_port=socks_port,
            connection_timeout=connection_timeout,
            stream_isolation=stream_isolation,
            nick_auth_mode=nick_auth_mode,
            nick_auth_directory_ids=nick_auth_directory_ids,
        )

        # Taker-specific state below.
        self.nick = nick_identity.nick
        self.neutrino_compat = neutrino_compat
        self.on_nick_leave = on_nick_leave

        # Direct peer connection settings
        self.prefer_direct_connections = prefer_direct_connections
        self.our_location = our_location
        # Peer connections indexed by nick
        self._peer_connections: dict[str, OnionPeer] = {}
        # Background tasks for pending connections
        self._pending_connect_tasks: dict[str, asyncio.Task[bool]] = {}

        # Unified message queue for direct peer messages
        # Messages from direct peers are queued here and consumed by wait_for_responses
        self._direct_message_queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue()

        # Multi-directory nick tracking
        # Format: active_nicks[nick] = {server1: True, server2: True, ...}
        # True = nick is present on this server, False = gone from this server
        # A nick is only considered completely gone when ALL servers report False
        self._active_nicks: dict[str, dict[str, bool]] = {}

    def _build_client_kwargs(self, host: str, port: int) -> dict[str, Any]:
        """Inject the taker's ``neutrino_compat`` flag into the base kwargs."""
        kwargs = super()._build_client_kwargs(host, port)
        kwargs["neutrino_compat"] = self.neutrino_compat
        return kwargs

    def _update_nick_status(self, nick: str, server: str, is_present: bool) -> None:
        """
        Update a nick's presence status on a specific directory server.

        If this causes the nick to become completely gone (absent from ALL servers),
        triggers the on_nick_leave callback.
        """
        if nick not in self._active_nicks:
            self._active_nicks[nick] = {}

        old_status = self._active_nicks[nick].get(server)
        self._active_nicks[nick][server] = is_present

        # Check if this update causes the nick to be completely gone
        if not is_present and old_status is True:
            # Nick just disappeared from this directory
            # Check if it's still present on any other directory
            if not any(status for status in self._active_nicks[nick].values()):
                logger.info(
                    f"Nick {nick} has left all directories "
                    f"(servers: {list(self._active_nicks[nick].keys())})"
                )
                if self.on_nick_leave:
                    self.on_nick_leave(nick)
                # Clean up the entry
                del self._active_nicks[nick]
        elif is_present and old_status is False:
            logger.debug(f"Nick {nick} returned to server {server}")

    def is_nick_active(self, nick: str) -> bool:
        """
        Check if a nick is active on at least one directory server.

        Returns:
            True if nick is present on at least one server
        """
        if nick not in self._active_nicks:
            return False
        return any(status for status in self._active_nicks[nick].values())

    def sync_nicks_with_peerlist(self, server: str, active_nicks: set[str]) -> None:
        """
        Synchronize nick tracking with a directory's peerlist.

        This is called after fetching a peerlist from a directory to update
        the nick tracking state. Nicks not in the peerlist are marked as gone
        from that directory.

        Args:
            server: The server identifier reporting the peerlist
            active_nicks: Set of nicks currently active on this server
        """
        # Mark all nicks in the peerlist as present
        for nick in active_nicks:
            self._update_nick_status(nick, server, True)

        # Mark nicks we're tracking but not in this peerlist as gone from this server
        for nick in list(self._active_nicks.keys()):
            if server in self._active_nicks[nick] and nick not in active_nicks:
                self._update_nick_status(nick, server, False)

    # =========================================================================
    # Direct Peer Connection Methods
    # =========================================================================

    def _get_peer_location(self, nick: str) -> str | None:
        """
        Get a maker's onion location from the peerlist.

        Args:
            nick: Maker's JoinMarket nick

        Returns:
            Onion address (host:port) or None if not found/not serving
        """
        for client in self.clients.values():
            location = client._active_peers.get(nick)
            if location and location != NOT_SERVING_ONION_HOSTNAME:
                return location
        return None

    def _should_try_direct_connect(self, nick: str) -> bool:
        """
        Check if we should attempt a direct connection to this peer.

        Returns False if:
        - Direct connections are disabled
        - We already have a connected peer
        - Peer doesn't serve an onion address
        - Connection attempt is already in progress
        """
        if not self.prefer_direct_connections:
            return False

        # Already connected?
        if nick in self._peer_connections:
            peer = self._peer_connections[nick]
            if peer.is_connected() or peer.is_connecting():
                return False

        # Connection attempt in progress?
        if nick in self._pending_connect_tasks:
            task = self._pending_connect_tasks[nick]
            if not task.done():
                return False

        # Has a valid onion address?
        location = self._get_peer_location(nick)
        return location is not None

    def _get_connected_peer(self, nick: str) -> OnionPeer | None:
        """
        Get a connected peer by nick.

        Returns:
            OnionPeer if connected and handshaked, None otherwise
        """
        peer = self._peer_connections.get(nick)
        if peer and peer.is_connected():
            return peer
        return None

    def bind_session(self, nick: str) -> ChannelBinding | None:
        """
        Pick and pin a single transport channel for a session with ``nick``.

        Encapsulates the channel-selection algorithm that callers
        previously open-coded by reaching into private attributes
        (``_get_connected_peer``, ``_active_nicks``, ``clients``,
        ``_active_peers``). The order is:

        1. If a direct peer connection is established and direct
           connections are preferred, bind to ``"direct"``.
        2. Otherwise, prefer a directory server that our nick-tracking
           layer marks as currently carrying ``nick``.
        3. Otherwise, fall back to any directory whose peerlist lists the
           nick.
        4. Otherwise, fall back to an arbitrary connected directory.

        Returns ``None`` only when no directories are connected, in which
        case the caller cannot send anything anyway.

        The returned :class:`ChannelBinding` should be reused for every
        subsequent message to ``nick`` in the same session: makers reject
        sessions whose ``!fill`` and ``!auth`` arrive via different
        channels.
        """
        peer = self._get_connected_peer(nick)
        peer_location = self._get_peer_location(nick)
        if peer is not None and self.prefer_direct_connections:
            return ChannelBinding(
                nick=nick,
                channel_id="direct",
                peer_location=peer_location,
            )

        # Prefer directories that have explicitly tracked this nick as active.
        target_directories: list[str] = []
        if nick in self._active_nicks:
            for server, is_active in self._active_nicks[nick].items():
                if is_active and server in self.clients:
                    target_directories.append(server)

        # Fall back to directories whose peerlist lists this nick.
        if not target_directories:
            for server, client in self.clients.items():
                if nick in client._active_peers:
                    target_directories.append(server)

        # Last-resort fallback: any connected directory.
        if not target_directories:
            target_directories = list(self.clients.keys())

        if not target_directories:
            return None

        chosen = target_directories[0]
        return ChannelBinding(
            nick=nick,
            channel_id=f"directory:{chosen}",
            peer_location=peer_location,
        )

    def upgrade_channel_prefer_direct(self, nick: str, current_channel: str) -> str:
        """Opportunistically upgrade a session channel to a direct connection.

        A session pins one channel before sending ``!fill`` (see
        :meth:`bind_session`). When ``!fill`` is sent via a directory while a
        direct connection is still being established, that connection often
        finishes handshaking before the taker sends ``!auth``/``!tx``. This
        mirrors the reference taker, which routes each privmsg
        opportunistically (``jmdaemon/onionmc.py::_privmsg``): once the direct
        peer is handshaked, later messages travel directly.

        Makers accept such mid-session directory->direct switches (the signed
        ``hostid`` is the fixed ``onion-network`` for all onion transports), so
        upgrading improves privacy and latency without breaking compatibility.

        We only ever upgrade directory->direct, never the reverse: if the
        session is already direct we keep it, and we never downgrade a direct
        session to a directory relay here.

        Args:
            nick: The maker nick whose session channel to re-evaluate.
            current_channel: The channel currently pinned for the session
                (``"direct"`` or ``"directory:<server>"``).

        Returns:
            ``"direct"`` if a handshaked direct connection is now available,
            otherwise ``current_channel`` unchanged.
        """
        if not self.prefer_direct_connections:
            return current_channel
        if current_channel == "direct":
            return current_channel
        if self._get_connected_peer(nick) is not None:
            logger.debug(
                f"Upgrading session channel for {nick} from "
                f"'{current_channel}' to 'direct' (direct connection now ready)"
            )
            return "direct"
        return current_channel

    def try_direct_connect(self, nick: str) -> None:
        """
        Public alias for ``_try_direct_connect``.

        Kicks off an opportunistic background connection attempt to
        ``nick``. Safe to call repeatedly: the underlying method is a
        no-op when a connection (or pending task) already exists.
        """
        self._try_direct_connect(nick)

    def get_peer_location(self, nick: str) -> str | None:
        """Public alias for ``_get_peer_location``."""
        return self._get_peer_location(nick)

    def get_connected_peer(self, nick: str) -> OnionPeer | None:
        """Public alias for ``_get_connected_peer``."""
        return self._get_connected_peer(nick)

    def get_pending_connect_task(self, nick: str) -> asyncio.Task[bool] | None:
        """Return the in-flight direct-connect task for ``nick``, if any."""
        return self._pending_connect_tasks.get(nick)

    async def _on_peer_message(self, nick: str, data: bytes) -> None:
        """
        Handle message received from a direct peer connection.

        Messages are forwarded to the unified direct message queue for processing
        by wait_for_responses(). The message is enriched with the sender's nick
        to match the format expected by the response processing logic.
        """
        try:
            import json

            msg = json.loads(data.decode("utf-8"))
            logger.debug(f"Received direct message from {nick}: type={msg.get('type')}")

            # Enrich message with sender nick for wait_for_responses to identify
            msg["from_nick"] = nick
            msg["from_direct"] = True

            # Queue for processing by wait_for_responses
            await self._direct_message_queue.put(msg)
        except Exception as e:
            logger.warning(f"Error processing peer message from {nick}: {e}")

    async def _on_peer_disconnect(self, nick: str) -> None:
        """Handle peer disconnection."""
        logger.debug(f"Peer {nick} disconnected")
        # Clean up but don't remove from _peer_connections immediately
        # in case we want to reconnect

    async def _on_peer_handshake_complete(self, nick: str) -> None:
        """Handle successful peer handshake."""
        logger.info(f"Direct connection established with {nick}")

    def _try_direct_connect(self, nick: str) -> None:
        """
        Opportunistically try to establish a direct connection to a maker.

        This is called asynchronously when sending a message via directory relay.
        The connection attempt runs in the background and future messages will
        use the direct connection if it succeeds.
        """
        if not self._should_try_direct_connect(nick):
            return

        location = self._get_peer_location(nick)
        if not location:
            return

        # Create peer if needed
        if nick not in self._peer_connections:
            peer = OnionPeer(
                nick=nick,
                location=location,
                socks_host=self.socks_host,
                socks_port=self.socks_port,
                timeout=self.connection_timeout,
                on_message=self._on_peer_message,
                on_disconnect=self._on_peer_disconnect,
                on_handshake_complete=self._on_peer_handshake_complete,
                nick_identity=self.nick_identity,
                socks_username=self._peer_creds[0],
                socks_password=self._peer_creds[1],
            )
            self._peer_connections[nick] = peer
        else:
            peer = self._peer_connections[nick]

        # Start connection in background
        task = peer.try_to_connect(
            our_nick=self.nick,
            our_location=self.our_location,
            network=self.network,
        )
        if task:
            self._pending_connect_tasks[nick] = task
            logger.debug(f"Started background connection to {nick} at {location}")

    async def _cleanup_peer_connections(self) -> None:
        """Clean up all peer connections (called on close)."""
        # Cancel pending connection tasks
        for nick, task in self._pending_connect_tasks.items():
            if not task.done():
                task.cancel()
        self._pending_connect_tasks.clear()

        # Disconnect all peers
        for nick, peer in self._peer_connections.items():
            try:
                await peer.disconnect()
            except Exception as e:
                logger.debug(f"Error disconnecting from peer {nick}: {e}")
        self._peer_connections.clear()

    async def connect_all(self) -> int:
        """Connect to all directory servers in parallel.

        Thin compatibility wrapper around
        :meth:`DirectoryClientPool.connect_all_parallel` that preserves
        the historical name used by the taker codebase.
        """
        return await self.connect_all_parallel()

    async def close_all(self) -> None:
        """Close all directory and peer connections.

        Peer (direct onion) connections are torn down first so any
        outgoing per-peer messages have a chance to flush before we
        close the relay channels. Directory client teardown is handled
        by :meth:`DirectoryClientPool.close_all`.
        """
        await self._cleanup_peer_connections()
        await super().close_all()

    async def fetch_orderbook(
        self,
        max_wait: float = 120.0,
        min_wait: float = 30.0,
        quiet_period: float = 15.0,
    ) -> list[Offer]:
        """
        Fetch orderbook from all connected directory servers in parallel.

        Trusts the directory's orderbook as authoritative - if a maker has an offer
        in the directory, they are considered online. This avoids incorrectly filtering
        offers as "stale" based on slow peerlist responses.

        Args:
            max_wait: Hard ceiling in seconds (default: 120s).
            min_wait: Minimum seconds before early exit is allowed (default: 30s).
            quiet_period: Seconds of silence before exiting early (default: 15s).
        """
        offers_by_key: dict[tuple[str, int], Offer] = {}

        async def fetch_from_server(
            server: str, client: DirectoryClient
        ) -> tuple[str, list[Offer]]:
            """Fetch offers from a single directory server."""
            try:
                offers, _bonds = await client.fetch_orderbooks(
                    max_wait=max_wait, min_wait=min_wait, quiet_period=quiet_period
                )
                return (server, offers)
            except Exception as e:
                logger.warning(f"Failed to fetch orderbook from {server}: {e}")
                return (server, [])

        # Fetch from all directories in parallel
        tasks = [fetch_from_server(server, client) for server, client in self.clients.items()]
        results = await asyncio.gather(*tasks)

        # Aggregate and deduplicate offers. A delayed directory response must
        # not let an older certificate suppress a renewal for the same offer.
        for server, offers in results:
            for offer in offers:
                key = (offer.counterparty, offer.oid)
                existing = offers_by_key.get(key)
                if existing is None:
                    offers_by_key[key] = offer
                    continue
                existing_expiry = (existing.fidelity_bond_data or {}).get("cert_expiry", -1)
                new_expiry = (offer.fidelity_bond_data or {}).get("cert_expiry", -1)
                if new_expiry > existing_expiry:
                    offers_by_key[key] = offer

        return list(offers_by_key.values())

    async def send_privmsg(
        self,
        recipient: str,
        command: str,
        data: str,
        log_routing: bool = False,
        force_channel: str | None = None,
    ) -> str:
        """Send a private message, respecting channel consistency for CoinJoin sessions.

        CRITICAL: Within a single CoinJoin session, all messages to a maker MUST use the
        same communication channel (either direct or a specific directory). Mixing channels
        causes the maker to reject messages as they appear to be from different sessions.

        Message routing priority (when force_channel is None):
        1. Direct peer connection (if connected and prefer_direct_connections=True)
        2. Directory relay (fallback)

        Args:
            recipient: Target maker nick
            command: Command name (without ! prefix)
            data: Command arguments
            log_routing: If True, log detailed routing information
            force_channel: If set, only use this channel:
                - "direct" = peer-to-peer onion connection
                - "directory:<host>:<port>" = relay through specific directory

        Returns:
            Channel used: "direct" or "directory:<host>:<port>"
        """
        # Get maker's direct onion location if available
        maker_location = self._get_peer_location(recipient)

        # If force_channel is set, use only that channel
        if force_channel:
            if force_channel == "direct":
                peer = self._get_connected_peer(recipient)
                if not peer:
                    raise RuntimeError(
                        f"Forced to use direct channel but no connection to {recipient}"
                    )
                success = await peer.send_privmsg(self.nick, command, data)
                if not success:
                    raise RuntimeError(f"Failed to send to {recipient} via direct connection")
                if log_routing:
                    logger.debug(
                        f"Sent !{command} to {recipient} via DIRECT connection "
                        f"(onion: {maker_location})"
                    )
                return "direct"
            elif force_channel.startswith("directory:"):
                # Extract host:port from "directory:host:port"
                server = force_channel[10:]  # Skip "directory:"
                client = self.clients.get(server)
                if not client:
                    raise RuntimeError(f"Forced to use directory {server} but not connected")
                await client.send_private_message(recipient, command, data)
                if log_routing:
                    logger.debug(
                        f"Sent !{command} to {recipient} via directory {server} "
                        f"(maker onion: {maker_location}, using relay)"
                    )
                return force_channel
            else:
                raise ValueError(f"Invalid force_channel: {force_channel}")

        # No forced channel - choose best available
        # Try direct connection first if available
        if self.prefer_direct_connections:
            peer = self._get_connected_peer(recipient)
            if peer:
                try:
                    success = await peer.send_privmsg(self.nick, command, data)
                    if success:
                        if log_routing:
                            logger.debug(
                                f"Sent !{command} to {recipient} via DIRECT connection "
                                f"(onion: {maker_location})"
                            )
                        return "direct"
                except Exception as e:
                    logger.debug(f"Direct send to {recipient} failed: {e}")

        # Fall back to directory relay
        # Opportunistically start direct connection for future messages
        if self.prefer_direct_connections and maker_location:
            self._try_direct_connect(recipient)

        # Identify valid directories for this recipient
        target_directories = []

        # Check active nicks tracking first
        if recipient in self._active_nicks:
            for server, is_active in self._active_nicks[recipient].items():
                if is_active and server in self.clients:
                    target_directories.append(server)

        # If not found in tracking (e.g. startup race), try all clients that list the peer
        if not target_directories:
            for server, client in self.clients.items():
                if recipient in client._active_peers:
                    target_directories.append(server)

        # If still not found, fall back to all connected clients (broadcast)
        if not target_directories:
            target_directories = list(self.clients.keys())

        # Shuffle to load balance
        secure_random.shuffle(target_directories)

        # Send via the first working directory
        # We strictly send to ONE directory to avoid message duplication
        for server in target_directories:
            client = self.clients.get(server)
            if not client:
                continue

            try:
                await client.send_private_message(recipient, command, data)
                if log_routing:
                    directory = f"{client.host}:{client.port}"
                    if maker_location:
                        logger.debug(
                            f"Sent !{command} to {recipient} via directory {directory} "
                            f"(maker onion: {maker_location}, using relay)"
                        )
                    else:
                        logger.debug(f"Sent !{command} to {recipient} via directory {directory}")
                # Success - return the channel used
                return f"directory:{server}"
            except Exception as e:
                logger.warning(f"Failed to send privmsg via {server}: {e}")

        raise RuntimeError(f"Failed to send !{command} to {recipient} via any directory")

    async def wait_for_responses(
        self,
        expected_nicks: list[str],
        expected_command: str,
        timeout: float = 60.0,
        expected_counts: dict[str, int] | None = None,
    ) -> dict[str, dict[str, Any]]:
        """Wait for responses from multiple makers at once.

        Listens for responses from BOTH:
        - Directory server message streams (via client.listen_for_messages())
        - Direct peer connections (via self._direct_message_queue)

        Returns a dict of nick -> response data for all makers that responded.
        Responses can include:
        - Normal responses matching expected_command
        - Error responses marked with "error": True

        Error handling:
        - Makers may send !error messages instead of the expected response
        - These indicate protocol failures (e.g., blacklisted PoDLE commitment)
        - Errors are returned in the response dict with {"error": True, "data": "reason"}

        Deduplication:
        - When connected to multiple directory servers, the same response may arrive
          multiple times. ResponseDeduplicator tracks which responses we've seen
          and logs duplicates for debugging.

        Special handling for !sig:
        - Makers send multiple !sig messages (one per UTXO)
        - We accumulate all messages in a list instead of keeping just the last one
        - Use expected_counts to specify how many signatures to expect per maker
        - Returns as soon as all expected signatures are received

        Args:
            expected_nicks: List of maker nicks to expect responses from
            expected_command: Command to wait for (e.g., "!pubkey", "!sig")
            timeout: Maximum time to wait in seconds
            expected_counts: For !sig, dict of nick -> expected signature count
        """
        # Track if this command expects multiple messages per maker
        accumulate_responses = expected_command == "!sig"

        responses: dict[str, dict[str, Any]] = {}
        remaining_nicks = set(expected_nicks)
        deduplicator = ResponseDeduplicator()
        # For !sig accumulation: track seen data per nick to drop cross-directory
        # duplicates (the same signature relayed by multiple directory servers).
        seen_sig_data: dict[str, set[str]] = {}
        loop = asyncio.get_event_loop()
        start_time = loop.time()
        # Servers whose listen failures were already reported at warning level;
        # subsequent failures are logged at debug to avoid flooding the log.
        listen_errors_reported: set[str] = set()

        def is_complete() -> bool:
            """Check if we have all expected responses."""
            if remaining_nicks:
                return False
            if accumulate_responses and expected_counts:
                # For !sig, check if we have all expected signatures
                for nick, expected in expected_counts.items():
                    if nick not in responses:
                        return False
                    received = len(responses[nick].get("data", []))
                    if received < expected:
                        return False
            return True

        def process_message(msg: dict[str, Any], source: str) -> None:
            """Process a single message from any source (directory or direct)."""
            nonlocal responses, remaining_nicks

            line = msg.get("line", "")
            if not line:
                return

            # Attribute strictly to the authenticated sender. Substring matching
            # let any peer claim another maker's nick by embedding it in payload.
            parsed = parse_jm_message(line)
            if parsed is None:
                return
            from_nick = parsed[0]
            if from_nick not in expected_nicks:
                return

            ok, command, _data = verify_signed_privmsg(from_nick, parsed[2], ONION_HOSTID)
            if not ok:
                logger.warning(f"Dropping unverified message from {from_nick} via {source}")
                return

            # Preserve the downstream payload contract (data plus pubkey/sig suffix).
            payload = parsed[2].split(" ", 1)[1].strip() if " " in parsed[2] else ""

            if command == "error":
                if not deduplicator.add_response(from_nick, "error", line, source):
                    return
                responses[from_nick] = {"error": True, "data": _data or "Unknown error"}
                remaining_nicks.discard(from_nick)
                logger.warning(f"Received error from {from_nick}: {_data}")
                return

            if command != expected_command.lstrip("!"):
                return

            if not accumulate_responses:
                if not deduplicator.add_response(from_nick, expected_command, line, source):
                    logger.debug(f"Duplicate {expected_command} from {from_nick} via {source}")
                    return
                responses[from_nick] = {"data": payload}
                remaining_nicks.discard(from_nick)
                logger.debug(f"Received {expected_command} from {from_nick} via {source}")
            else:
                # Accumulate !sig messages, deduplicating identical content relayed
                # by multiple directories.
                nick_seen = seen_sig_data.setdefault(from_nick, set())
                if payload in nick_seen:
                    return
                nick_seen.add(payload)
                if from_nick not in responses:
                    responses[from_nick] = {"data": []}
                    remaining_nicks.discard(from_nick)
                responses[from_nick]["data"].append(payload)
                logger.debug(
                    f"Received {expected_command} #{len(responses[from_nick]['data'])} "
                    f"from {from_nick} via {source}"
                )

        while not is_complete():
            elapsed = loop.time() - start_time
            if elapsed >= timeout:
                if not accumulate_responses:
                    logger.warning(
                        f"Timeout waiting for {expected_command} from: {remaining_nicks}"
                    )
                elif expected_counts:
                    # Log which makers haven't sent all signatures
                    for nick, expected in expected_counts.items():
                        received = len(responses.get(nick, {}).get("data", []))
                        if received < expected:
                            logger.warning(f"Timeout: {nick} sent {received}/{expected} signatures")
                break

            remaining_time = min(5.0, timeout - elapsed)  # Listen in 5s chunks

            # First, drain any pending direct peer messages (non-blocking)
            while True:
                try:
                    msg = self._direct_message_queue.get_nowait()
                    process_message(msg, "direct")
                except asyncio.QueueEmpty:
                    break

            # Check if we have everything after processing direct messages
            if is_complete():
                break

            # Listen to all directory clients concurrently for shorter duration
            # Use 1s chunks to allow more frequent checking of direct message queue
            listen_duration = min(1.0, remaining_time)

            async def listen_to_client(
                server: str, client: DirectoryClient
            ) -> list[tuple[str, dict[str, Any]]]:
                try:
                    messages = await client.listen_for_messages(duration=listen_duration)
                    return [(server, msg) for msg in messages]
                except Exception as e:
                    if server not in listen_errors_reported:
                        listen_errors_reported.add(server)
                        logger.warning(f"Error listening to {server}: {e}")
                    else:
                        logger.debug(f"Error listening to {server}: {e}")
                    return []

            # Gather messages from all directories concurrently
            listen_started = loop.time()
            results = await asyncio.gather(
                *[listen_to_client(s, c) for s, c in self.clients.items()]
            )
            for result_list in results:
                for server, msg in result_list:
                    process_message(msg, f"directory:{server}")

            # Pace the loop: when every directory listen fails immediately
            # (e.g. all connections closed) or there are no directory clients
            # at all, the gather above returns in microseconds. Without a
            # sleep this degenerates into a busy loop that spins thousands of
            # iterations per second until the timeout, flooding the log.
            listen_elapsed = loop.time() - listen_started
            if listen_elapsed < listen_duration and not is_complete():
                await asyncio.sleep(listen_duration - listen_elapsed)

        # Log deduplication stats if there were duplicates
        stats = deduplicator.stats
        if stats.duplicates_dropped > 0:
            logger.debug(
                f"Response deduplication: {stats.unique_messages} unique, "
                f"{stats.duplicates_dropped} duplicates dropped "
                f"({stats.duplicate_rate:.1f}% duplicate rate)"
            )

        return responses

    async def wait_for_response(
        self,
        from_nick: str,
        expected_command: str,
        timeout: float = 30.0,
    ) -> dict[str, Any] | None:
        """Wait for a specific response from a maker (legacy method)."""
        responses = await self.wait_for_responses([from_nick], expected_command, timeout)
        return responses.get(from_nick)

Attributes

neutrino_compat = neutrino_compat instance-attribute

nick = nick_identity.nick instance-attribute

on_nick_leave = on_nick_leave instance-attribute

our_location = our_location instance-attribute

prefer_direct_connections = prefer_direct_connections instance-attribute

Methods:

__init__(directory_servers: list[str], network: str, nick_identity: NickIdentity, socks_host: str = '127.0.0.1', socks_port: int = 9050, connection_timeout: float = 120.0, neutrino_compat: bool = False, on_nick_leave: Any | None = None, prefer_direct_connections: bool = True, our_location: str = 'NOT-SERVING-ONION', stream_isolation: bool = False, nick_auth_mode: NickAuthMode = NickAuthMode.PREFER_VERIFIED, nick_auth_directory_ids: dict[str, str] | None = None)

Source code in taker/src/taker/multi_directory.py

def __init__(
    self,
    directory_servers: list[str],
    network: str,
    nick_identity: NickIdentity,
    socks_host: str = "127.0.0.1",
    socks_port: int = 9050,
    connection_timeout: float = 120.0,
    neutrino_compat: bool = False,
    on_nick_leave: Any | None = None,
    prefer_direct_connections: bool = True,
    our_location: str = "NOT-SERVING-ONION",
    stream_isolation: bool = False,
    nick_auth_mode: NickAuthMode = NickAuthMode.PREFER_VERIFIED,
    nick_auth_directory_ids: dict[str, str] | None = None,
):
    # Connection / SOCKS / credential setup is delegated to the
    # DirectoryClientPool base; it handles directory_servers, network,
    # nick_identity, SOCKS params, connection_timeout, stream_isolation,
    # the clients dict, and _dir_creds / _peer_creds.
    super().__init__(
        directory_servers=directory_servers,
        network=network,
        nick_identity=nick_identity,
        socks_host=socks_host,
        socks_port=socks_port,
        connection_timeout=connection_timeout,
        stream_isolation=stream_isolation,
        nick_auth_mode=nick_auth_mode,
        nick_auth_directory_ids=nick_auth_directory_ids,
    )

    # Taker-specific state below.
    self.nick = nick_identity.nick
    self.neutrino_compat = neutrino_compat
    self.on_nick_leave = on_nick_leave

    # Direct peer connection settings
    self.prefer_direct_connections = prefer_direct_connections
    self.our_location = our_location
    # Peer connections indexed by nick
    self._peer_connections: dict[str, OnionPeer] = {}
    # Background tasks for pending connections
    self._pending_connect_tasks: dict[str, asyncio.Task[bool]] = {}

    # Unified message queue for direct peer messages
    # Messages from direct peers are queued here and consumed by wait_for_responses
    self._direct_message_queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue()

    # Multi-directory nick tracking
    # Format: active_nicks[nick] = {server1: True, server2: True, ...}
    # True = nick is present on this server, False = gone from this server
    # A nick is only considered completely gone when ALL servers report False
    self._active_nicks: dict[str, dict[str, bool]] = {}

bind_session(nick: str) -> ChannelBinding | None

Pick and pin a single transport channel for a session with nick.

Encapsulates the channel-selection algorithm that callers previously open-coded by reaching into private attributes (_get_connected_peer, _active_nicks, clients, _active_peers). The order is:

1. If a direct peer connection is established and direct connections are preferred, bind to "direct".
2. Otherwise, prefer a directory server that our nick-tracking layer marks as currently carrying nick.
3. Otherwise, fall back to any directory whose peerlist lists the nick.
4. Otherwise, fall back to an arbitrary connected directory.

Returns None only when no directories are connected, in which case the caller cannot send anything anyway.

The returned :class:ChannelBinding should be reused for every subsequent message to nick in the same session: makers reject sessions whose !fill and !auth arrive via different channels.

Source code in taker/src/taker/multi_directory.py

def bind_session(self, nick: str) -> ChannelBinding | None:
    """
    Pick and pin a single transport channel for a session with ``nick``.

    Encapsulates the channel-selection algorithm that callers
    previously open-coded by reaching into private attributes
    (``_get_connected_peer``, ``_active_nicks``, ``clients``,
    ``_active_peers``). The order is:

    1. If a direct peer connection is established and direct
       connections are preferred, bind to ``"direct"``.
    2. Otherwise, prefer a directory server that our nick-tracking
       layer marks as currently carrying ``nick``.
    3. Otherwise, fall back to any directory whose peerlist lists the
       nick.
    4. Otherwise, fall back to an arbitrary connected directory.

    Returns ``None`` only when no directories are connected, in which
    case the caller cannot send anything anyway.

    The returned :class:`ChannelBinding` should be reused for every
    subsequent message to ``nick`` in the same session: makers reject
    sessions whose ``!fill`` and ``!auth`` arrive via different
    channels.
    """
    peer = self._get_connected_peer(nick)
    peer_location = self._get_peer_location(nick)
    if peer is not None and self.prefer_direct_connections:
        return ChannelBinding(
            nick=nick,
            channel_id="direct",
            peer_location=peer_location,
        )

    # Prefer directories that have explicitly tracked this nick as active.
    target_directories: list[str] = []
    if nick in self._active_nicks:
        for server, is_active in self._active_nicks[nick].items():
            if is_active and server in self.clients:
                target_directories.append(server)

    # Fall back to directories whose peerlist lists this nick.
    if not target_directories:
        for server, client in self.clients.items():
            if nick in client._active_peers:
                target_directories.append(server)

    # Last-resort fallback: any connected directory.
    if not target_directories:
        target_directories = list(self.clients.keys())

    if not target_directories:
        return None

    chosen = target_directories[0]
    return ChannelBinding(
        nick=nick,
        channel_id=f"directory:{chosen}",
        peer_location=peer_location,
    )

close_all() -> None async

Close all directory and peer connections.

Peer (direct onion) connections are torn down first so any outgoing per-peer messages have a chance to flush before we close the relay channels. Directory client teardown is handled by :meth:DirectoryClientPool.close_all.

Source code in taker/src/taker/multi_directory.py

async def close_all(self) -> None:
    """Close all directory and peer connections.

    Peer (direct onion) connections are torn down first so any
    outgoing per-peer messages have a chance to flush before we
    close the relay channels. Directory client teardown is handled
    by :meth:`DirectoryClientPool.close_all`.
    """
    await self._cleanup_peer_connections()
    await super().close_all()

connect_all() -> int async

Connect to all directory servers in parallel.

Thin compatibility wrapper around :meth:DirectoryClientPool.connect_all_parallel that preserves the historical name used by the taker codebase.

Source code in taker/src/taker/multi_directory.py

async def connect_all(self) -> int:
    """Connect to all directory servers in parallel.

    Thin compatibility wrapper around
    :meth:`DirectoryClientPool.connect_all_parallel` that preserves
    the historical name used by the taker codebase.
    """
    return await self.connect_all_parallel()

fetch_orderbook(max_wait: float = 120.0, min_wait: float = 30.0, quiet_period: float = 15.0) -> list[Offer] async

Fetch orderbook from all connected directory servers in parallel.

Trusts the directory's orderbook as authoritative - if a maker has an offer in the directory, they are considered online. This avoids incorrectly filtering offers as "stale" based on slow peerlist responses.

Args: max_wait: Hard ceiling in seconds (default: 120s). min_wait: Minimum seconds before early exit is allowed (default: 30s). quiet_period: Seconds of silence before exiting early (default: 15s).

Source code in taker/src/taker/multi_directory.py

async def fetch_orderbook(
    self,
    max_wait: float = 120.0,
    min_wait: float = 30.0,
    quiet_period: float = 15.0,
) -> list[Offer]:
    """
    Fetch orderbook from all connected directory servers in parallel.

    Trusts the directory's orderbook as authoritative - if a maker has an offer
    in the directory, they are considered online. This avoids incorrectly filtering
    offers as "stale" based on slow peerlist responses.

    Args:
        max_wait: Hard ceiling in seconds (default: 120s).
        min_wait: Minimum seconds before early exit is allowed (default: 30s).
        quiet_period: Seconds of silence before exiting early (default: 15s).
    """
    offers_by_key: dict[tuple[str, int], Offer] = {}

    async def fetch_from_server(
        server: str, client: DirectoryClient
    ) -> tuple[str, list[Offer]]:
        """Fetch offers from a single directory server."""
        try:
            offers, _bonds = await client.fetch_orderbooks(
                max_wait=max_wait, min_wait=min_wait, quiet_period=quiet_period
            )
            return (server, offers)
        except Exception as e:
            logger.warning(f"Failed to fetch orderbook from {server}: {e}")
            return (server, [])

    # Fetch from all directories in parallel
    tasks = [fetch_from_server(server, client) for server, client in self.clients.items()]
    results = await asyncio.gather(*tasks)

    # Aggregate and deduplicate offers. A delayed directory response must
    # not let an older certificate suppress a renewal for the same offer.
    for server, offers in results:
        for offer in offers:
            key = (offer.counterparty, offer.oid)
            existing = offers_by_key.get(key)
            if existing is None:
                offers_by_key[key] = offer
                continue
            existing_expiry = (existing.fidelity_bond_data or {}).get("cert_expiry", -1)
            new_expiry = (offer.fidelity_bond_data or {}).get("cert_expiry", -1)
            if new_expiry > existing_expiry:
                offers_by_key[key] = offer

    return list(offers_by_key.values())

get_connected_peer(nick: str) -> OnionPeer | None

Public alias for _get_connected_peer.

Source code in taker/src/taker/multi_directory.py

def get_connected_peer(self, nick: str) -> OnionPeer | None:
    """Public alias for ``_get_connected_peer``."""
    return self._get_connected_peer(nick)

get_peer_location(nick: str) -> str | None

Public alias for _get_peer_location.

Source code in taker/src/taker/multi_directory.py

def get_peer_location(self, nick: str) -> str | None:
    """Public alias for ``_get_peer_location``."""
    return self._get_peer_location(nick)

get_pending_connect_task(nick: str) -> asyncio.Task[bool] | None

Return the in-flight direct-connect task for nick, if any.

Source code in taker/src/taker/multi_directory.py

def get_pending_connect_task(self, nick: str) -> asyncio.Task[bool] | None:
    """Return the in-flight direct-connect task for ``nick``, if any."""
    return self._pending_connect_tasks.get(nick)

is_nick_active(nick: str) -> bool

Check if a nick is active on at least one directory server.

Returns: True if nick is present on at least one server

Source code in taker/src/taker/multi_directory.py

def is_nick_active(self, nick: str) -> bool:
    """
    Check if a nick is active on at least one directory server.

    Returns:
        True if nick is present on at least one server
    """
    if nick not in self._active_nicks:
        return False
    return any(status for status in self._active_nicks[nick].values())

send_privmsg(recipient: str, command: str, data: str, log_routing: bool = False, force_channel: str | None = None) -> str async

Send a private message, respecting channel consistency for CoinJoin sessions.

CRITICAL: Within a single CoinJoin session, all messages to a maker MUST use the same communication channel (either direct or a specific directory). Mixing channels causes the maker to reject messages as they appear to be from different sessions.

Message routing priority (when force_channel is None): 1. Direct peer connection (if connected and prefer_direct_connections=True) 2. Directory relay (fallback)

Args: recipient: Target maker nick command: Command name (without ! prefix) data: Command arguments log_routing: If True, log detailed routing information force_channel: If set, only use this channel: - "direct" = peer-to-peer onion connection - "directory::" = relay through specific directory

Returns: Channel used: "direct" or "directory::"

Source code in taker/src/taker/multi_directory.py

async def send_privmsg(
    self,
    recipient: str,
    command: str,
    data: str,
    log_routing: bool = False,
    force_channel: str | None = None,
) -> str:
    """Send a private message, respecting channel consistency for CoinJoin sessions.

    CRITICAL: Within a single CoinJoin session, all messages to a maker MUST use the
    same communication channel (either direct or a specific directory). Mixing channels
    causes the maker to reject messages as they appear to be from different sessions.

    Message routing priority (when force_channel is None):
    1. Direct peer connection (if connected and prefer_direct_connections=True)
    2. Directory relay (fallback)

    Args:
        recipient: Target maker nick
        command: Command name (without ! prefix)
        data: Command arguments
        log_routing: If True, log detailed routing information
        force_channel: If set, only use this channel:
            - "direct" = peer-to-peer onion connection
            - "directory:<host>:<port>" = relay through specific directory

    Returns:
        Channel used: "direct" or "directory:<host>:<port>"
    """
    # Get maker's direct onion location if available
    maker_location = self._get_peer_location(recipient)

    # If force_channel is set, use only that channel
    if force_channel:
        if force_channel == "direct":
            peer = self._get_connected_peer(recipient)
            if not peer:
                raise RuntimeError(
                    f"Forced to use direct channel but no connection to {recipient}"
                )
            success = await peer.send_privmsg(self.nick, command, data)
            if not success:
                raise RuntimeError(f"Failed to send to {recipient} via direct connection")
            if log_routing:
                logger.debug(
                    f"Sent !{command} to {recipient} via DIRECT connection "
                    f"(onion: {maker_location})"
                )
            return "direct"
        elif force_channel.startswith("directory:"):
            # Extract host:port from "directory:host:port"
            server = force_channel[10:]  # Skip "directory:"
            client = self.clients.get(server)
            if not client:
                raise RuntimeError(f"Forced to use directory {server} but not connected")
            await client.send_private_message(recipient, command, data)
            if log_routing:
                logger.debug(
                    f"Sent !{command} to {recipient} via directory {server} "
                    f"(maker onion: {maker_location}, using relay)"
                )
            return force_channel
        else:
            raise ValueError(f"Invalid force_channel: {force_channel}")

    # No forced channel - choose best available
    # Try direct connection first if available
    if self.prefer_direct_connections:
        peer = self._get_connected_peer(recipient)
        if peer:
            try:
                success = await peer.send_privmsg(self.nick, command, data)
                if success:
                    if log_routing:
                        logger.debug(
                            f"Sent !{command} to {recipient} via DIRECT connection "
                            f"(onion: {maker_location})"
                        )
                    return "direct"
            except Exception as e:
                logger.debug(f"Direct send to {recipient} failed: {e}")

    # Fall back to directory relay
    # Opportunistically start direct connection for future messages
    if self.prefer_direct_connections and maker_location:
        self._try_direct_connect(recipient)

    # Identify valid directories for this recipient
    target_directories = []

    # Check active nicks tracking first
    if recipient in self._active_nicks:
        for server, is_active in self._active_nicks[recipient].items():
            if is_active and server in self.clients:
                target_directories.append(server)

    # If not found in tracking (e.g. startup race), try all clients that list the peer
    if not target_directories:
        for server, client in self.clients.items():
            if recipient in client._active_peers:
                target_directories.append(server)

    # If still not found, fall back to all connected clients (broadcast)
    if not target_directories:
        target_directories = list(self.clients.keys())

    # Shuffle to load balance
    secure_random.shuffle(target_directories)

    # Send via the first working directory
    # We strictly send to ONE directory to avoid message duplication
    for server in target_directories:
        client = self.clients.get(server)
        if not client:
            continue

        try:
            await client.send_private_message(recipient, command, data)
            if log_routing:
                directory = f"{client.host}:{client.port}"
                if maker_location:
                    logger.debug(
                        f"Sent !{command} to {recipient} via directory {directory} "
                        f"(maker onion: {maker_location}, using relay)"
                    )
                else:
                    logger.debug(f"Sent !{command} to {recipient} via directory {directory}")
            # Success - return the channel used
            return f"directory:{server}"
        except Exception as e:
            logger.warning(f"Failed to send privmsg via {server}: {e}")

    raise RuntimeError(f"Failed to send !{command} to {recipient} via any directory")

sync_nicks_with_peerlist(server: str, active_nicks: set[str]) -> None

Synchronize nick tracking with a directory's peerlist.

This is called after fetching a peerlist from a directory to update the nick tracking state. Nicks not in the peerlist are marked as gone from that directory.

Args: server: The server identifier reporting the peerlist active_nicks: Set of nicks currently active on this server

Source code in taker/src/taker/multi_directory.py

def sync_nicks_with_peerlist(self, server: str, active_nicks: set[str]) -> None:
    """
    Synchronize nick tracking with a directory's peerlist.

    This is called after fetching a peerlist from a directory to update
    the nick tracking state. Nicks not in the peerlist are marked as gone
    from that directory.

    Args:
        server: The server identifier reporting the peerlist
        active_nicks: Set of nicks currently active on this server
    """
    # Mark all nicks in the peerlist as present
    for nick in active_nicks:
        self._update_nick_status(nick, server, True)

    # Mark nicks we're tracking but not in this peerlist as gone from this server
    for nick in list(self._active_nicks.keys()):
        if server in self._active_nicks[nick] and nick not in active_nicks:
            self._update_nick_status(nick, server, False)

try_direct_connect(nick: str) -> None

Public alias for _try_direct_connect.

Kicks off an opportunistic background connection attempt to nick. Safe to call repeatedly: the underlying method is a no-op when a connection (or pending task) already exists.

Source code in taker/src/taker/multi_directory.py

def try_direct_connect(self, nick: str) -> None:
    """
    Public alias for ``_try_direct_connect``.

    Kicks off an opportunistic background connection attempt to
    ``nick``. Safe to call repeatedly: the underlying method is a
    no-op when a connection (or pending task) already exists.
    """
    self._try_direct_connect(nick)

upgrade_channel_prefer_direct(nick: str, current_channel: str) -> str

Opportunistically upgrade a session channel to a direct connection.

A session pins one channel before sending !fill (see :meth:bind_session). When !fill is sent via a directory while a direct connection is still being established, that connection often finishes handshaking before the taker sends !auth/!tx. This mirrors the reference taker, which routes each privmsg opportunistically (jmdaemon/onionmc.py::_privmsg): once the direct peer is handshaked, later messages travel directly.

Makers accept such mid-session directory->direct switches (the signed hostid is the fixed onion-network for all onion transports), so upgrading improves privacy and latency without breaking compatibility.

We only ever upgrade directory->direct, never the reverse: if the session is already direct we keep it, and we never downgrade a direct session to a directory relay here.

Args: nick: The maker nick whose session channel to re-evaluate. current_channel: The channel currently pinned for the session ("direct" or "directory:<server>").

Returns: "direct" if a handshaked direct connection is now available, otherwise current_channel unchanged.

Source code in taker/src/taker/multi_directory.py

def upgrade_channel_prefer_direct(self, nick: str, current_channel: str) -> str:
    """Opportunistically upgrade a session channel to a direct connection.

    A session pins one channel before sending ``!fill`` (see
    :meth:`bind_session`). When ``!fill`` is sent via a directory while a
    direct connection is still being established, that connection often
    finishes handshaking before the taker sends ``!auth``/``!tx``. This
    mirrors the reference taker, which routes each privmsg
    opportunistically (``jmdaemon/onionmc.py::_privmsg``): once the direct
    peer is handshaked, later messages travel directly.

    Makers accept such mid-session directory->direct switches (the signed
    ``hostid`` is the fixed ``onion-network`` for all onion transports), so
    upgrading improves privacy and latency without breaking compatibility.

    We only ever upgrade directory->direct, never the reverse: if the
    session is already direct we keep it, and we never downgrade a direct
    session to a directory relay here.

    Args:
        nick: The maker nick whose session channel to re-evaluate.
        current_channel: The channel currently pinned for the session
            (``"direct"`` or ``"directory:<server>"``).

    Returns:
        ``"direct"`` if a handshaked direct connection is now available,
        otherwise ``current_channel`` unchanged.
    """
    if not self.prefer_direct_connections:
        return current_channel
    if current_channel == "direct":
        return current_channel
    if self._get_connected_peer(nick) is not None:
        logger.debug(
            f"Upgrading session channel for {nick} from "
            f"'{current_channel}' to 'direct' (direct connection now ready)"
        )
        return "direct"
    return current_channel

wait_for_response(from_nick: str, expected_command: str, timeout: float = 30.0) -> dict[str, Any] | None async

Wait for a specific response from a maker (legacy method).

Source code in taker/src/taker/multi_directory.py

async def wait_for_response(
    self,
    from_nick: str,
    expected_command: str,
    timeout: float = 30.0,
) -> dict[str, Any] | None:
    """Wait for a specific response from a maker (legacy method)."""
    responses = await self.wait_for_responses([from_nick], expected_command, timeout)
    return responses.get(from_nick)

wait_for_responses(expected_nicks: list[str], expected_command: str, timeout: float = 60.0, expected_counts: dict[str, int] | None = None) -> dict[str, dict[str, Any]] async

Wait for responses from multiple makers at once.

Listens for responses from BOTH: - Directory server message streams (via client.listen_for_messages()) - Direct peer connections (via self._direct_message_queue)

Returns a dict of nick -> response data for all makers that responded. Responses can include: - Normal responses matching expected_command - Error responses marked with "error": True

Error handling: - Makers may send !error messages instead of the expected response - These indicate protocol failures (e.g., blacklisted PoDLE commitment) - Errors are returned in the response dict with {"error": True, "data": "reason"}

Deduplication: - When connected to multiple directory servers, the same response may arrive multiple times. ResponseDeduplicator tracks which responses we've seen and logs duplicates for debugging.

Special handling for !sig: - Makers send multiple !sig messages (one per UTXO) - We accumulate all messages in a list instead of keeping just the last one - Use expected_counts to specify how many signatures to expect per maker - Returns as soon as all expected signatures are received

Args: expected_nicks: List of maker nicks to expect responses from expected_command: Command to wait for (e.g., "!pubkey", "!sig") timeout: Maximum time to wait in seconds expected_counts: For !sig, dict of nick -> expected signature count

Source code in taker/src/taker/multi_directory.py

async def wait_for_responses(
    self,
    expected_nicks: list[str],
    expected_command: str,
    timeout: float = 60.0,
    expected_counts: dict[str, int] | None = None,
) -> dict[str, dict[str, Any]]:
    """Wait for responses from multiple makers at once.

    Listens for responses from BOTH:
    - Directory server message streams (via client.listen_for_messages())
    - Direct peer connections (via self._direct_message_queue)

    Returns a dict of nick -> response data for all makers that responded.
    Responses can include:
    - Normal responses matching expected_command
    - Error responses marked with "error": True

    Error handling:
    - Makers may send !error messages instead of the expected response
    - These indicate protocol failures (e.g., blacklisted PoDLE commitment)
    - Errors are returned in the response dict with {"error": True, "data": "reason"}

    Deduplication:
    - When connected to multiple directory servers, the same response may arrive
      multiple times. ResponseDeduplicator tracks which responses we've seen
      and logs duplicates for debugging.

    Special handling for !sig:
    - Makers send multiple !sig messages (one per UTXO)
    - We accumulate all messages in a list instead of keeping just the last one
    - Use expected_counts to specify how many signatures to expect per maker
    - Returns as soon as all expected signatures are received

    Args:
        expected_nicks: List of maker nicks to expect responses from
        expected_command: Command to wait for (e.g., "!pubkey", "!sig")
        timeout: Maximum time to wait in seconds
        expected_counts: For !sig, dict of nick -> expected signature count
    """
    # Track if this command expects multiple messages per maker
    accumulate_responses = expected_command == "!sig"

    responses: dict[str, dict[str, Any]] = {}
    remaining_nicks = set(expected_nicks)
    deduplicator = ResponseDeduplicator()
    # For !sig accumulation: track seen data per nick to drop cross-directory
    # duplicates (the same signature relayed by multiple directory servers).
    seen_sig_data: dict[str, set[str]] = {}
    loop = asyncio.get_event_loop()
    start_time = loop.time()
    # Servers whose listen failures were already reported at warning level;
    # subsequent failures are logged at debug to avoid flooding the log.
    listen_errors_reported: set[str] = set()

    def is_complete() -> bool:
        """Check if we have all expected responses."""
        if remaining_nicks:
            return False
        if accumulate_responses and expected_counts:
            # For !sig, check if we have all expected signatures
            for nick, expected in expected_counts.items():
                if nick not in responses:
                    return False
                received = len(responses[nick].get("data", []))
                if received < expected:
                    return False
        return True

    def process_message(msg: dict[str, Any], source: str) -> None:
        """Process a single message from any source (directory or direct)."""
        nonlocal responses, remaining_nicks

        line = msg.get("line", "")
        if not line:
            return

        # Attribute strictly to the authenticated sender. Substring matching
        # let any peer claim another maker's nick by embedding it in payload.
        parsed = parse_jm_message(line)
        if parsed is None:
            return
        from_nick = parsed[0]
        if from_nick not in expected_nicks:
            return

        ok, command, _data = verify_signed_privmsg(from_nick, parsed[2], ONION_HOSTID)
        if not ok:
            logger.warning(f"Dropping unverified message from {from_nick} via {source}")
            return

        # Preserve the downstream payload contract (data plus pubkey/sig suffix).
        payload = parsed[2].split(" ", 1)[1].strip() if " " in parsed[2] else ""

        if command == "error":
            if not deduplicator.add_response(from_nick, "error", line, source):
                return
            responses[from_nick] = {"error": True, "data": _data or "Unknown error"}
            remaining_nicks.discard(from_nick)
            logger.warning(f"Received error from {from_nick}: {_data}")
            return

        if command != expected_command.lstrip("!"):
            return

        if not accumulate_responses:
            if not deduplicator.add_response(from_nick, expected_command, line, source):
                logger.debug(f"Duplicate {expected_command} from {from_nick} via {source}")
                return
            responses[from_nick] = {"data": payload}
            remaining_nicks.discard(from_nick)
            logger.debug(f"Received {expected_command} from {from_nick} via {source}")
        else:
            # Accumulate !sig messages, deduplicating identical content relayed
            # by multiple directories.
            nick_seen = seen_sig_data.setdefault(from_nick, set())
            if payload in nick_seen:
                return
            nick_seen.add(payload)
            if from_nick not in responses:
                responses[from_nick] = {"data": []}
                remaining_nicks.discard(from_nick)
            responses[from_nick]["data"].append(payload)
            logger.debug(
                f"Received {expected_command} #{len(responses[from_nick]['data'])} "
                f"from {from_nick} via {source}"
            )

    while not is_complete():
        elapsed = loop.time() - start_time
        if elapsed >= timeout:
            if not accumulate_responses:
                logger.warning(
                    f"Timeout waiting for {expected_command} from: {remaining_nicks}"
                )
            elif expected_counts:
                # Log which makers haven't sent all signatures
                for nick, expected in expected_counts.items():
                    received = len(responses.get(nick, {}).get("data", []))
                    if received < expected:
                        logger.warning(f"Timeout: {nick} sent {received}/{expected} signatures")
            break

        remaining_time = min(5.0, timeout - elapsed)  # Listen in 5s chunks

        # First, drain any pending direct peer messages (non-blocking)
        while True:
            try:
                msg = self._direct_message_queue.get_nowait()
                process_message(msg, "direct")
            except asyncio.QueueEmpty:
                break

        # Check if we have everything after processing direct messages
        if is_complete():
            break

        # Listen to all directory clients concurrently for shorter duration
        # Use 1s chunks to allow more frequent checking of direct message queue
        listen_duration = min(1.0, remaining_time)

        async def listen_to_client(
            server: str, client: DirectoryClient
        ) -> list[tuple[str, dict[str, Any]]]:
            try:
                messages = await client.listen_for_messages(duration=listen_duration)
                return [(server, msg) for msg in messages]
            except Exception as e:
                if server not in listen_errors_reported:
                    listen_errors_reported.add(server)
                    logger.warning(f"Error listening to {server}: {e}")
                else:
                    logger.debug(f"Error listening to {server}: {e}")
                return []

        # Gather messages from all directories concurrently
        listen_started = loop.time()
        results = await asyncio.gather(
            *[listen_to_client(s, c) for s, c in self.clients.items()]
        )
        for result_list in results:
            for server, msg in result_list:
                process_message(msg, f"directory:{server}")

        # Pace the loop: when every directory listen fails immediately
        # (e.g. all connections closed) or there are no directory clients
        # at all, the gather above returns in microseconds. Without a
        # sleep this degenerates into a busy loop that spins thousands of
        # iterations per second until the timeout, flooding the log.
        listen_elapsed = loop.time() - listen_started
        if listen_elapsed < listen_duration and not is_complete():
            await asyncio.sleep(listen_duration - listen_elapsed)

    # Log deduplication stats if there were duplicates
    stats = deduplicator.stats
    if stats.duplicates_dropped > 0:
        logger.debug(
            f"Response deduplication: {stats.unique_messages} unique, "
            f"{stats.duplicates_dropped} duplicates dropped "
            f"({stats.duplicate_rate:.1f}% duplicate rate)"
        )

    return responses

PhaseResult

Result from a CoinJoin phase with failed maker tracking.

Used to communicate phase outcomes and enable maker replacement logic.

Source code in taker/src/taker/models.py

@dataclass
class PhaseResult:
    """Result from a CoinJoin phase with failed maker tracking.

    Used to communicate phase outcomes and enable maker replacement logic.
    """

    success: bool
    failed_makers: list[str] = Field(default_factory=list)
    blacklist_error: bool = False  # True if any maker rejected due to blacklisted commitment
    # Subset of failed_makers that specifically rejected with a "blacklist" error.
    # Used so the taker can tell "minority blacklist rejection" (probably a lying
    # or out-of-sync maker -> ignore + replace the maker) from "majority blacklist
    # rejection" (commitment really is known -> rotate commitment).
    blacklist_makers: list[str] = Field(default_factory=list)

    @property
    def needs_replacement(self) -> bool:
        """True if phase failed due to non-responsive makers (not other errors)."""
        return not self.success and len(self.failed_makers) > 0

Attributes

blacklist_error: bool = False class-attribute instance-attribute

blacklist_makers: list[str] = Field(default_factory=list) class-attribute instance-attribute

failed_makers: list[str] = Field(default_factory=list) class-attribute instance-attribute

needs_replacement: bool property

True if phase failed due to non-responsive makers (not other errors).

success: bool instance-attribute

Taker

Bases: TakerMonitoringMixin

Main Taker class for executing CoinJoin transactions.

Source code in taker/src/taker/taker.py

class Taker(TakerMonitoringMixin):
    """
    Main Taker class for executing CoinJoin transactions.
    """

    def __init__(
        self,
        wallet: WalletService,
        backend: BlockchainBackend,
        config: TakerConfig,
        confirmation_callback: Any | None = None,
    ):
        """
        Initialize the Taker.

        Args:
            wallet: Wallet service for UTXO management and signing
            backend: Blockchain backend for broadcasting
            config: Taker configuration
            confirmation_callback: Optional callback for user confirmation before proceeding
        """
        self.wallet = wallet
        self.backend = backend
        self.config = config
        self.confirmation_callback = confirmation_callback

        self.nick_identity = NickIdentity(JM_VERSION)
        self.nick = self.nick_identity.nick
        self.state = TakerState.IDLE
        # Source mixdepth of the most recent ``do_coinjoin`` call. With
        # interactive selection this is derived from the selected UTXOs, so
        # callers (e.g. the CLI confirmation prompt) can display it.
        self.last_source_mixdepth: int | None = None

        # Advertise neutrino_compat if our backend can provide extended UTXO metadata.
        # This tells other peers that we can provide scriptpubkey and blockheight.
        # Full nodes (Bitcoin Core) can provide this; light clients (Neutrino) cannot.
        neutrino_compat = backend.can_provide_neutrino_metadata()

        # Directory client
        self.directory_client = MultiDirectoryClient(
            directory_servers=config.directory_servers,
            network=config.network.value,
            nick_identity=self.nick_identity,
            socks_host=config.socks_host,
            socks_port=config.socks_port,
            connection_timeout=config.connection_timeout,
            nick_auth_mode=config.nick_auth_mode,
            nick_auth_directory_ids=config.nick_auth_directory_ids,
            neutrino_compat=neutrino_compat,
            stream_isolation=config.stream_isolation,
        )

        # Orderbook manager
        # Read maker nick from state file to exclude from peer selection (self-CoinJoin protection)
        own_wallet_nicks: set[str] = set()
        maker_nick = read_nick_state(config.data_dir, "maker")
        if maker_nick:
            own_wallet_nicks.add(maker_nick)
            logger.info(f"Self-CoinJoin protection: excluding maker nick {maker_nick}")

        self.orderbook_manager = OrderbookManager(
            config.max_cj_fee,
            bondless_makers_allowance=config.bondless_makers_allowance,
            bondless_require_zero_fee=config.bondless_makers_allowance_require_zero_fee,
            data_dir=config.data_dir,
            own_wallet_nicks=own_wallet_nicks,
        )

        # PoDLE manager for commitment tracking
        self.podle_manager = PoDLEManager(config.data_dir)

        # Per-CoinJoin state and protocol phases live in a dedicated
        # ``CoinJoinSession``. ``Taker`` owns persistent infrastructure
        # (wallet, backend, config, directory client, orderbook manager,
        # PoDLE manager, schedule) and delegates the protocol phases to the
        # session, which is reset at the start of each ``do_coinjoin`` call.
        self._session = CoinJoinSession()
        self._session.attach(self)
        self._round_lock = asyncio.Lock()

        # Schedule for tumbler-style operations
        self.schedule: Schedule | None = None

        # Background task tracking
        self.running = False
        self._background_tasks: list[asyncio.Task[None]] = []

    async def sync_wallet(self) -> int:
        """
        Sync the wallet and return total balance.

        This method is separated from start() to allow callers to check
        funds before connecting to directory servers (avoiding unnecessary
        network connections when funds are insufficient).

        Returns:
            Total wallet balance in satoshis.
        """
        logger.info(f"Starting taker (nick: {self.nick})")

        # Log wallet name if using descriptor wallet backend
        from jmwallet.backends.descriptor_wallet import DescriptorWalletBackend

        if isinstance(self.backend, DescriptorWalletBackend):
            logger.info(f"Using wallet: {self.backend.wallet_name}")

        # Initialize commitment blacklist with configured data directory
        set_blacklist_path(data_dir=self.config.data_dir)

        # Sync wallet
        logger.info("Syncing wallet...")

        # Setup descriptor wallet if needed (one-time operation)
        if isinstance(self.backend, DescriptorWalletBackend):
            if not await self.wallet.is_descriptor_wallet_ready():
                logger.info("Descriptor wallet not set up. Importing descriptors...")
                await self.wallet.setup_descriptor_wallet(rescan=True)
                logger.info("Descriptor wallet setup complete")

            # Use fast descriptor wallet sync
            await self.wallet.sync_with_descriptor_wallet()
        else:
            # Use standard sync (BIP157/158 for neutrino, mempool API, etc.)
            await self.wallet.sync_all()
        await self.wallet.reconstruct_imported_state_safe()

        total_balance = await self.wallet.get_total_balance()
        logger.info(f"Wallet synced. Total balance: {total_balance:,} sats")

        return total_balance

    async def connect(self) -> None:
        """
        Connect to directory servers and start background tasks.

        This should be called after sync_wallet() and any fund validation.
        """
        # Connect to directory servers
        logger.info("Connecting to directory servers...")
        connected = await self.directory_client.connect_all()

        if connected == 0:
            raise RuntimeError("Failed to connect to any directory server")

        logger.info(f"Connected to {connected} directory servers")

        # Mark as running and start background tasks
        self.running = True

        # Start pending transaction monitor
        monitor_task = asyncio.create_task(self._monitor_pending_transactions())
        self._background_tasks.append(monitor_task)

        # Start periodic rescan task (useful for schedule mode)
        rescan_task = asyncio.create_task(self._periodic_rescan())
        self._background_tasks.append(rescan_task)

        # Start periodic directory connection status logging task
        conn_status_task = asyncio.create_task(self._periodic_directory_connection_status())
        self._background_tasks.append(conn_status_task)

    async def start(self) -> None:
        """
        Start the taker: sync wallet and connect to directory servers.

        This is a convenience method that calls sync_wallet() followed by connect().
        For early fund validation, call sync_wallet() first, validate, then call connect().
        """
        await self.sync_wallet()
        await self.connect()

    def release_input_locks(self) -> None:
        """Clean up persisted CoinJoin locks held on this round's taker inputs.

        Pre-sign failures release immediately. Once signatures may exist, the
        owner-qualified leases are renewed through the pending window instead.
        """
        if self._session and self._session.signing_boundary_crossed:
            self._session.retain_input_locks()
            return
        if self._session and self._session.reserved_inputs:
            try:
                self.wallet.release_coinjoin_inputs(
                    self._session.reserved_inputs,
                    owner=self._session.input_lock_owner,
                )
            except Exception as e:  # pragma: no cover - best-effort cleanup
                logger.debug(f"Failed to release taker input locks: {e}")
            self._session.reserved_inputs = set()

    @property
    def last_failure_reason(self) -> str | None:
        """Reason the most recent ``do_coinjoin`` call failed (or ``None``).

        Forwarded from the per-round :class:`CoinJoinSession` so external
        consumers (e.g. the tumbler runner) can surface why a round did not
        broadcast without reaching into private session state.
        """
        return self._session.last_failure_reason

    @property
    def last_used_nicks(self) -> set[str]:
        """Maker nicks used by the most recent ``do_coinjoin`` call."""
        return self._session.last_used_nicks

    async def check_utxo_eligibility(self, amount: int, mixdepth: int | None) -> str | None:
        """Validate that ``mixdepth`` can fund a CoinJoin of ``amount``.

        Runs the same eligibility filters used later in :meth:`do_coinjoin`
        (confirmations, frozen, fidelity bonds, in-flight locks, the mixdepth-0
        merge restriction and the PoDLE size requirement) *before* any network
        operation, so an ineligible wallet fails fast instead of after a long
        directory/orderbook/bond cycle (issue #528).

        Args:
            amount: Target amount in satoshis (``0`` for sweep).
            mixdepth: Source mixdepth. ``None`` is only meaningful with
                interactive selection (``select_utxos``), where it means "any
                mixdepth" (the source is derived from the selection later);
                otherwise it falls back to mixdepth 0.

        Returns:
            ``None`` when a CoinJoin can proceed, otherwise a human-readable
            reason describing why it cannot.
        """
        min_conf = self.config.taker_utxo_age

        # Interactive selection follows different rules (the user may pick
        # unlocked fidelity bonds and is not bound to auto-selection), so only
        # require that *something* is selectable here.
        if self.config.select_utxos:
            mixdepths = (
                [mixdepth] if mixdepth is not None else list(range(self.wallet.mixdepth_count))
            )
            utxos = []
            for md in mixdepths:
                utxos.extend(await self.wallet.get_utxos(md))
            reserved = self.wallet.get_locked_input_outpoints()
            if not selectable_for_interactive(utxos, min_conf, excluded_outpoints=reserved):
                if mixdepth is not None:
                    return classify_utxos(
                        utxos, mixdepth, min_conf, reserved_outpoints=reserved
                    ).no_eligible_reason()
                return (
                    "No selectable UTXOs in any mixdepth (all UTXOs are "
                    "frozen, immature, locked fidelity bonds, or in use)"
                )
            return None

        if mixdepth is None:
            mixdepth = 0
        utxos = await self.wallet.get_utxos(mixdepth)

        reserved = self.wallet.get_locked_input_outpoints()
        breakdown = classify_utxos(utxos, mixdepth, min_conf, reserved_outpoints=reserved)

        if not breakdown.eligible:
            return breakdown.no_eligible_reason()

        # Sweep spends every eligible UTXO; a non-empty pool is sufficient.
        is_sweep = amount == 0
        if is_sweep:
            return None

        # PoDLE necessary condition: a commitment needs a UTXO worth at least
        # ``taker_utxo_amtpercent`` of the amount. Without one the round always
        # fails at commitment generation, so reject early with a clear message.
        if not podle_threshold_met(
            breakdown.eligible, amount, min_conf, self.config.taker_utxo_amtpercent
        ):
            min_value = int(amount * self.config.taker_utxo_amtpercent / 100)
            return (
                f"No eligible UTXO in mixdepth {mixdepth} is large enough for the "
                f"PoDLE commitment: need at least {min_value:,} sats "
                f"({self.config.taker_utxo_amtpercent}% of {amount:,} sats, "
                f"taker_utxo_amtpercent). Use a larger UTXO or lower the amount."
            )

        # Amount coverage: dry-run the exact selection used later so the verdict
        # matches reality (including the mixdepth-0 merge restriction).
        try:
            self.wallet.select_utxos(
                mixdepth,
                amount,
                min_conf,
                exclude=reserved,
            )
        except ValueError as exc:
            return _append_confirmation_hint(str(exc), min_conf)

        return None

    async def stop(self, *, close_wallet: bool = True) -> None:
        """Stop the taker and close connections.

        Args:
            close_wallet: If ``True`` (the default), also close the wallet's
                backend connection. Pass ``False`` when the wallet is shared
                with another component (e.g. a jmwalletd tumbler runner that
                will reuse the same :class:`~jmwallet.wallet.service.WalletService`
                instance across multiple taker phases) to avoid tearing down a
                still-in-use wallet.
        """
        logger.info("Stopping taker...")
        self.running = False

        # Cancel all background tasks
        for task in self._background_tasks:
            task.cancel()

        if self._background_tasks:
            await asyncio.gather(*self._background_tasks, return_exceptions=True)
        self._background_tasks.clear()

        await self.directory_client.close_all()
        if close_wallet:
            await self.wallet.close()
        logger.info("Taker stopped")

    async def _update_offers_with_bond_values(self, offers: list[Offer]) -> None:
        """
        Verify fidelity bonds and calculate their values.

        Uses the backend's ``verify_bonds()`` method for efficient bulk verification
        that works correctly on all backends (Bitcoin Core, neutrino, mempool).

        For each offer with a fidelity bond proof, derives the P2WSH bond address
        from the UTXO public key and locktime, then delegates verification to the
        backend which can batch the lookups optimally.
        """
        for offer in offers:
            offer.fidelity_bond_value = 0

        bonded_offers = [offer for offer in offers if offer.fidelity_bond_data]
        if not bonded_offers:
            return

        try:
            current_block_height = await self.backend.get_block_height()
        except Exception as e:
            logger.warning(f"Cannot verify fidelity bond certificate expiry: {e}")
            return
        if type(current_block_height) is not int or current_block_height < 0:
            logger.warning(
                f"Cannot verify fidelity bond certificate expiry: backend returned "
                f"invalid block height {current_block_height!r}"
            )
            return

        # Deduplicate identical claims while verifying conflicting script claims
        # independently. A claim is the outpoint plus its proof-derived script.
        claim_to_request: dict[tuple[str, int, str], BondVerificationRequest] = {}
        claim_to_offers: dict[tuple[str, int, str], list[Offer]] = {}

        for offer in bonded_offers:
            bond_data = offer.fidelity_bond_data
            assert bond_data is not None

            txid = bond_data["utxo_txid"]
            vout = bond_data["utxo_vout"]
            cert_expiry_height = bond_data.get("cert_expiry")

            if not isinstance(cert_expiry_height, int):
                logger.debug(f"Bond {txid}:{vout} missing certificate expiry, skipping")
                continue
            if current_block_height > cert_expiry_height:
                logger.debug(
                    f"Bond {txid}:{vout} certificate expired at block "
                    f"{cert_expiry_height} (current block {current_block_height})"
                )
                continue

            locktime = bond_data["locktime"]
            utxo_pub = bond_data.get("utxo_pub")

            if not utxo_pub:
                logger.debug(f"Bond {txid}:{vout} missing utxo_pub, skipping")
                continue

            try:
                utxo_pub_bytes = bytes.fromhex(utxo_pub) if isinstance(utxo_pub, str) else utxo_pub
                bond_addr = derive_bond_address(utxo_pub_bytes, locktime, self.config.network)
            except Exception as e:
                logger.debug(f"Failed to derive bond address for {txid}:{vout}: {e}")
                continue

            request = BondVerificationRequest(
                txid=txid,
                vout=vout,
                utxo_pub=utxo_pub_bytes,
                locktime=locktime,
                address=bond_addr.address,
                scriptpubkey=bond_addr.scriptpubkey.hex(),
            )
            claim_key = (txid, vout, request.scriptpubkey)
            if claim_key in claim_to_request:
                claim_to_offers[claim_key].append(offer)
                continue

            claim_to_request[claim_key] = request
            claim_to_offers[claim_key] = [offer]

        if not claim_to_request:
            return

        logger.info(f"Verifying {len(claim_to_request)} fidelity bonds...")

        # Bulk verify via the backend (batched for efficiency)
        try:
            requests = list(claim_to_request.values())
            results = await self.backend.verify_bonds(requests)
        except Exception as e:
            logger.warning(f"Bond verification failed: {e}")
            return
        if len(results) != len(requests):
            logger.warning(
                f"Bond verification returned {len(results)} results for {len(requests)} requests"
            )
            return

        current_time = int(time.time())
        claim_values: dict[tuple[str, int, str], int] = {}

        for request, result in zip(requests, results, strict=True):
            if (result.txid, result.vout) != (request.txid, request.vout):
                logger.warning(
                    f"Bond verification result mismatch: requested "
                    f"{request.txid}:{request.vout}, received {result.txid}:{result.vout}"
                )
                continue
            if not result.valid:
                logger.debug(f"Bond {result.txid}:{result.vout} invalid: {result.error}")
                continue

            bond_value = calculate_timelocked_fidelity_bond_value(
                utxo_value=result.value,
                confirmation_time=result.block_time,
                locktime=request.locktime,
                current_time=current_time,
            )

            if bond_value > 0:
                claim_key = (request.txid, request.vout, request.scriptpubkey)
                claim_values[claim_key] = bond_value

        # Update only offers whose certificate and proof data were eligible.
        updated_count = 0
        for claim_key, bond_value in claim_values.items():
            for offer in claim_to_offers[claim_key]:
                offer.fidelity_bond_value = bond_value
                updated_count += 1

        logger.info(f"Updated {updated_count} offers with verified fidelity bond values")

    async def do_coinjoin(
        self,
        amount: int,
        destination: str,
        mixdepth: int | None = None,
        counterparty_count: int | None = None,
        exclude_nicks: set[str] | None = None,
    ) -> str | None:
        """Run one CoinJoin with fresh, non-reusable per-round state."""
        if self._round_lock.locked():
            logger.error("A CoinJoin round is already active on this Taker instance")
            return None

        async with self._round_lock:
            session = CoinJoinSession()
            session.attach(self)
            self._session = session
            self.state = TakerState.IDLE
            return await self._do_coinjoin(
                amount=amount,
                destination=destination,
                mixdepth=mixdepth,
                counterparty_count=counterparty_count,
                exclude_nicks=exclude_nicks,
            )

    async def _do_coinjoin(
        self,
        amount: int,
        destination: str,
        mixdepth: int | None = None,
        counterparty_count: int | None = None,
        exclude_nicks: set[str] | None = None,
    ) -> str | None:
        """
        Execute a single CoinJoin transaction.

        Args:
            amount: Amount in satoshis (0 for sweep)
            destination: Destination address ("INTERNAL" for next mixdepth)
            mixdepth: Source mixdepth. ``None`` means: derive it from the
                interactive UTXO selection when ``select_utxos`` is enabled
                (the first selected UTXO pins the mixdepth), otherwise fall
                back to mixdepth 0.
            counterparty_count: Number of makers (default from config)
            exclude_nicks: Additional maker nicks to exclude from selection
                (on top of ``orderbook_manager.ignored_makers`` and
                ``own_wallet_nicks``). Tumbler uses this to prevent the
                same maker from re-appearing across consecutive plan phases.

        Returns:
            Transaction ID if successful, None otherwise
        """
        try:
            # Reset per-call state so callers reading ``last_used_nicks`` after
            # a failure don't pick up nicks from a previous successful round.
            self._session.last_used_nicks = set()
            # When the caller does not pin a counterparty count, fall back to
            # the configured value (which may itself be ``None`` to request a
            # random draw from the upstream-aligned [8, 10] range).
            self._session.last_failure_reason = None

            # Re-read maker nick state on every coinjoin attempt.  The maker
            # may have been started *after* this Taker was constructed (common
            # in tumbler runs), so the nick read at __init__ time would be
            # stale.  Refreshing here ensures the hard exclusion is always
            # current regardless of startup order.
            current_maker_nick = read_nick_state(self.config.data_dir, "maker")
            if current_maker_nick:
                if current_maker_nick not in self.orderbook_manager.own_wallet_nicks:
                    logger.info(
                        f"Self-CoinJoin protection: adding maker nick {current_maker_nick} "
                        "to exclusion set (detected after taker init)"
                    )
                self.orderbook_manager.own_wallet_nicks.add(current_maker_nick)

            requested = (
                counterparty_count
                if counterparty_count is not None
                else self.config.counterparty_count
            )
            n_makers = resolve_counterparty_count(requested)

            # Interactive UTXO selection happens first (before any network
            # work): the selector shows the whole wallet and, unless the
            # caller pinned a mixdepth, the source mixdepth is derived from
            # the user's selection.
            manually_selected_utxos: list[UTXOInfo] | None = None
            if self.config.select_utxos:
                logger.info("Launching interactive UTXO selection...")
                manually_selected_utxos = await self._maybe_select_utxos_interactively(
                    amount=amount,
                    mixdepth=mixdepth,
                )
                if not manually_selected_utxos:
                    return None
                mixdepth = manually_selected_utxos[0].mixdepth
                logger.info(f"Source mixdepth: {mixdepth} (from selection)")
            elif mixdepth is None:
                mixdepth = 0
            self.last_source_mixdepth = mixdepth

            # Pre-flight: reject ineligible UTXOs before any orderbook/bond work
            # so the user is not kept waiting on a doomed round (issue #528).
            # Callers that already validate (the CLI) will simply re-confirm a
            # passing verdict; jmwalletd, run_schedule and the tumbler rely on
            # this check since they go straight to do_coinjoin.
            eligibility_reason = await self.check_utxo_eligibility(amount, mixdepth)
            if eligibility_reason is not None:
                logger.error(eligibility_reason)
                self._session.last_failure_reason = eligibility_reason
                self.state = TakerState.FAILED
                return None

            # Determine destination address
            if destination == "INTERNAL":
                dest_mixdepth = (mixdepth + 1) % self.wallet.mixdepth_count
                # Use internal chain (/1) for CoinJoin outputs, not external (/0)
                # This matches the reference implementation behavior where all JM-generated
                # addresses (CJ outputs and change) use the internal branch
                destination = self.wallet.get_new_internal_address(dest_mixdepth)
                logger.info(f"Using internal address: {destination}")
            else:
                # Warn when the user-supplied destination does not match the
                # wallet's native script type (#113).
                warn_if_destination_script_mismatch(destination)

            # Resolve fee rate early (before any fee estimation calls)
            try:
                await self._session._resolve_fee_rate()
            except ValueError as e:
                logger.error(str(e))
                self._session.last_failure_reason = str(e)
                self.state = TakerState.FAILED
                return None

            # Track if this is a sweep (no change) transaction
            self._session.is_sweep = amount == 0

            # UTXO selection (interactive or automatic) is done before fetching
            # the orderbook to avoid wasting the user's time on a doomed round.
            # Now fetch orderbook after UTXO selection is done
            self.state = TakerState.FETCHING_ORDERBOOK
            logger.info("Fetching orderbook...")
            offers = await self.directory_client.fetch_orderbook(
                max_wait=self.config.order_wait_time,
                min_wait=self.config.orderbook_min_wait,
                quiet_period=self.config.orderbook_quiet_period,
            )

            # Determine required features for maker selection.
            # Neutrino takers require makers that support extended UTXO metadata
            # (scriptPubKey + blockheight) via the neutrino_compat feature.
            required_features: set[str] | None = None
            if self.backend.requires_neutrino_metadata():
                required_features = {FEATURE_NEUTRINO_COMPAT}

            # Early compatibility pre-check for neutrino takers: count how many offers
            # are from makers known to support neutrino_compat (via peerlist_features or
            # the deprecated !neutrino flag). This lets us fail fast before the expensive
            # fidelity bond verification, which can take 20+ minutes on neutrino backends.
            #
            # Feature detection comes from two sources:
            # 1. peerlist_features: directories that support it report per-peer features
            # 2. !neutrino flag in offers (deprecated but still parsed)
            #
            # Offers with empty features dicts (unknown status) are NOT rejected here --
            # they pass through and will be verified during _phase_auth(). Only offers
            # where we KNOW the maker lacks the feature are filtered out.
            if required_features:
                known_compatible = sum(
                    1
                    for o in offers
                    if o.features.get(FEATURE_NEUTRINO_COMPAT) or o.neutrino_compat
                )
                known_incompatible = sum(
                    1
                    for o in offers
                    if o.features
                    and not o.features.get(FEATURE_NEUTRINO_COMPAT)
                    and not o.neutrino_compat
                )
                unknown = len(offers) - known_compatible - known_incompatible
                logger.info(
                    f"Neutrino compatibility pre-check: {known_compatible} compatible, "
                    f"{known_incompatible} incompatible, {unknown} unknown "
                    f"(from {len(offers)} total offers)"
                )

                # If even the most optimistic count (compatible + unknown) can't meet
                # the requirement, fail immediately before bond verification.
                if known_compatible + unknown < n_makers:
                    reason = (
                        f"Not enough potentially compatible makers for neutrino taker: "
                        f"need {n_makers}, but only {known_compatible} known compatible + "
                        f"{unknown} unknown = {known_compatible + unknown} possible. "
                        f"{known_incompatible} offers filtered as incompatible (no "
                        f"neutrino_compat). Bond verification skipped."
                    )
                    logger.error(reason)
                    self._session.last_failure_reason = reason
                    self.state = TakerState.FAILED
                    return None

                if known_compatible < n_makers and unknown > 0:
                    logger.warning(
                        f"Only {known_compatible} offers confirmed neutrino_compat, "
                        f"need {n_makers}. {unknown} offers have unknown feature status "
                        f"and will be checked during handshake. Not all directory servers "
                        f"support peerlist_features."
                    )

            # Verify and calculate fidelity bond values
            await self._update_offers_with_bond_values(offers)

            self.orderbook_manager.update_offers(offers)

            if len(offers) < n_makers:
                reason = f"Not enough offers: need {n_makers}, found {len(offers)}"
                logger.error(reason)
                self._session.last_failure_reason = reason
                self.state = TakerState.FAILED
                return None

            if required_features:
                logger.info(
                    "Neutrino backend: requiring neutrino_compat in offer filtering, "
                    "will also negotiate during handshake"
                )

            self.state = TakerState.SELECTING_MAKERS

            if self._session.is_sweep:
                # SWEEP MODE: Select ALL UTXOs and calculate exact cj_amount for zero change
                logger.info("Sweep mode: selecting UTXOs from mixdepth")

                # Use manually selected UTXOs if available, otherwise get all UTXOs
                if manually_selected_utxos:
                    self._session.preselected_utxos = manually_selected_utxos
                    logger.info(
                        f"Sweep using {len(manually_selected_utxos)} manually selected UTXOs "
                        f"(--select-utxos was used)"
                    )
                else:
                    # Get ALL UTXOs from the mixdepth (default sweep behavior)
                    locked_inputs = self.wallet.get_locked_input_outpoints()
                    self._session.preselected_utxos = self.wallet.get_all_utxos(
                        mixdepth,
                        self.config.taker_utxo_age,
                        exclude=locked_inputs,
                    )
                    logger.info(
                        f"Sweep using all {len(self._session.preselected_utxos)} UTXOs "
                        f"from mixdepth (no --select-utxos)"
                    )

                if not self._session.preselected_utxos:
                    reason = f"No eligible UTXOs in mixdepth {mixdepth}"
                    logger.error(reason)
                    self._session.last_failure_reason = reason
                    self.state = TakerState.FAILED
                    return None

                total_input_value = sum(u.value for u in self._session.preselected_utxos)
                logger.info(
                    f"Sweep: {len(self._session.preselected_utxos)} UTXOs, "
                    f"total value: {total_input_value:,} sats"
                )

                # Estimate tx fee for sweep order calculation
                # Conservative estimate: 2 inputs per maker + buffer for edge cases
                # Most makers have 1-2 inputs, but occasionally one might have 6+.
                # The buffer (5 inputs) covers the edge case without being excessive.
                # If actual < estimated: extra goes to miner (acceptable)
                # If actual > estimated: CoinJoin fails with negative residual error
                maker_inputs_per_maker = 2
                maker_inputs_buffer = 5  # Extra inputs to handle edge cases
                estimated_inputs = (
                    len(self._session.preselected_utxos)
                    + n_makers * maker_inputs_per_maker
                    + maker_inputs_buffer
                )
                # CJ outputs + maker changes (no taker change in sweep!)
                estimated_outputs = 1 + n_makers + n_makers
                # For sweeps, use base rate for deterministic budget calculation.
                # The cj_amount is calculated based on this budget, so it must match
                # exactly at build time. Using randomized rate would cause residual fees.
                estimated_tx_fee = self._session._estimate_tx_fee(
                    estimated_inputs, estimated_outputs, use_base_rate=True
                )

                # Store the tx fee budget for use at build time.
                # This is critical: the cj_amount is calculated based on this budget,
                # so we MUST use this same value at build time to avoid residual fees.
                self._session._sweep_tx_fee_budget = estimated_tx_fee

                # Use sweep order selection - this calculates exact cj_amount for zero change
                selected_offers, self._session.cj_amount, total_fee = (
                    self.orderbook_manager.select_makers_for_sweep(
                        total_input_value=total_input_value,
                        my_txfee=estimated_tx_fee,
                        n=n_makers,
                        required_features=required_features,
                        exclude_nicks=exclude_nicks,
                    )
                )

                if len(selected_offers) < self.config.minimum_makers:
                    reason = f"Not enough makers for sweep: {len(selected_offers)}"
                    logger.error(reason)
                    self._session.last_failure_reason = reason
                    self.state = TakerState.FAILED
                    return None

                logger.info(
                    f"Sweep: cj_amount={self._session.cj_amount:,} sats calculated for zero change"
                )
                # Record initial counterparties so callers (e.g. the tumbler)
                # can avoid reusing them in the next round, even if a
                # replacement maker is later swapped in.
                self._session.last_used_nicks = set(selected_offers.keys())

            else:
                # NORMAL MODE: Select minimum UTXOs needed
                self._session.cj_amount = amount
                logger.info(f"Selecting {n_makers} makers for {self._session.cj_amount:,} sats...")

                selected_offers, total_fee = self.orderbook_manager.select_makers(
                    cj_amount=self._session.cj_amount,
                    n=n_makers,
                    required_features=required_features,
                    exclude_nicks=exclude_nicks,
                )

                if len(selected_offers) < self.config.minimum_makers:
                    reason = f"Not enough makers selected: {len(selected_offers)}"
                    logger.error(reason)
                    self._session.last_failure_reason = reason
                    self.state = TakerState.FAILED
                    return None

                # Record initial counterparties so callers (e.g. the tumbler)
                # can avoid reusing them in the next round, even if a
                # replacement maker is later swapped in.
                self._session.last_used_nicks = set(selected_offers.keys())

                # Pre-select UTXOs for CoinJoin, then generate PoDLE from one of them
                # This ensures the PoDLE UTXO is one we'll actually use in the transaction
                logger.info("Selecting UTXOs and generating PoDLE commitment...")

                # Use manually selected UTXOs if available
                if manually_selected_utxos:
                    self._session.preselected_utxos = manually_selected_utxos
                    logger.info(
                        f"Using {len(manually_selected_utxos)} manually selected UTXOs "
                        f"(total: {sum(u.value for u in manually_selected_utxos):,} sats)"
                    )
                else:
                    # Estimate required amount (conservative estimate for UTXO pre-selection)
                    # We'll refine this in _phase_build_tx once we have exact maker UTXOs
                    estimated_inputs = 2 + len(selected_offers) * 2  # Rough estimate
                    estimated_outputs = 2 + len(selected_offers) * 2
                    estimated_tx_fee = self._session._estimate_tx_fee(
                        estimated_inputs, estimated_outputs
                    )
                    estimated_required = self._session.cj_amount + total_fee + estimated_tx_fee

                    # Pre-select UTXOs for the CoinJoin, skipping any inputs
                    # locked by another in-flight round (this or another process
                    # on the same wallet) so we don't build a conflicting tx.
                    locked_inputs = self.wallet.get_locked_input_outpoints()
                    try:
                        self._session.preselected_utxos = self.wallet.select_utxos(
                            mixdepth,
                            estimated_required,
                            self.config.taker_utxo_age,
                            exclude=locked_inputs,
                        )
                        preselected = self._session.preselected_utxos
                        logger.info(
                            f"Pre-selected {len(preselected)} UTXOs for CoinJoin "
                            f"(total: {sum(u.value for u in preselected):,} sats)"
                        )
                    except ValueError as e:
                        reason = _append_confirmation_hint(str(e), self.config.taker_utxo_age)
                        logger.error(reason)
                        self._session.last_failure_reason = reason
                        self.state = TakerState.FAILED
                        return None

            # "Block first, then continue": persist a lock on our chosen inputs
            # before negotiating with makers, so a concurrent round on the same
            # wallet cannot pick the same UTXO and produce a conflicting
            # transaction. The lock auto-expires (and is released on failure),
            # so a crash never blocks these funds permanently.
            to_reserve = {(u.txid, u.vout) for u in self._session.preselected_utxos}
            if not self.wallet.reserve_coinjoin_inputs(
                to_reserve,
                ttl=self._session.input_lock_ttl_sec(),
                owner=self._session.input_lock_owner,
            ):
                reason = (
                    "Selected UTXOs are locked by another in-flight CoinJoin on "
                    "this wallet (avoid running concurrent rounds on one wallet)."
                )
                logger.error(reason)
                self._session.last_failure_reason = reason
                self.state = TakerState.FAILED
                return None
            self._session.reserved_inputs |= to_reserve

            # Initialize maker sessions - neutrino_compat will be detected during handshake
            # when we receive the !pubkey response with features field
            self._session.maker_sessions = {
                nick: MakerSession(nick=nick, offer=offer, supports_neutrino_compat=False)
                for nick, offer in selected_offers.items()
            }

            logger.info(
                f"Selected {len(self._session.maker_sessions)} makers, "
                f"total fee: {total_fee:,} sats"
            )

            # Log estimated transaction fee before prompting for confirmation
            # Conservative estimate: assume 1 input per maker + 20% buffer, rounded up
            import math

            estimated_maker_inputs = math.ceil(n_makers * 1.2)
            estimated_inputs = len(self._session.preselected_utxos) + estimated_maker_inputs
            # Outputs: 1 CJ output per participant + change outputs (assume all have change)
            estimated_outputs = (1 + n_makers) + (1 + n_makers)
            estimated_tx_fee = self._session._estimate_tx_fee(estimated_inputs, estimated_outputs)
            logger.info(
                f"Estimated transaction (mining) fee: {estimated_tx_fee:,} sats "
                f"(~{self._session._fee_rate:.2f} sat/vB for ~{estimated_inputs} inputs, "
                f"{estimated_outputs} outputs)"
            )

            # Prompt for confirmation after maker selection
            if hasattr(self, "confirmation_callback") and self.confirmation_callback:
                try:
                    # Build maker details for confirmation
                    maker_details = []
                    for nick, session in self._session.maker_sessions.items():
                        fee = session.offer.calculate_fee(self._session.cj_amount)
                        bond_value = session.offer.fidelity_bond_value
                        # Get maker's location from any connected directory
                        location = None
                        for client in self.directory_client.clients.values():
                            location = client._active_peers.get(nick)
                            if location and location != "NOT-SERVING-ONION":
                                break
                        maker_details.append(
                            {
                                "nick": nick,
                                "fee": fee,
                                "bond_value": bond_value,
                                "location": location,
                            }
                        )

                    confirmed = self.confirmation_callback(
                        maker_details=maker_details,
                        cj_amount=self._session.cj_amount,
                        total_fee=total_fee + estimated_tx_fee,
                        destination=destination,
                        mining_fee=estimated_tx_fee,
                        fee_rate=self._session._fee_rate,
                        stage="initial",
                    )
                    if not confirmed:
                        logger.info("CoinJoin cancelled by user")
                        self.state = TakerState.CANCELLED
                        return None
                except Exception as e:
                    logger.error(f"Confirmation failed: {e}")
                    self.state = TakerState.FAILED
                    return None

            def get_private_key(addr: str) -> bytes | None:
                key = self.wallet.get_key_for_address(addr)
                if key is None:
                    return None
                return key.get_private_key_bytes()

            # Generate PoDLE from pre-selected UTXOs only
            # This ensures the commitment is from a UTXO that will be in the transaction
            self._session.podle_commitment = self.podle_manager.generate_fresh_commitment(
                wallet_utxos=self._session.preselected_utxos,  # Only from pre-selected UTXOs!
                cj_amount=self._session.cj_amount,
                private_key_getter=get_private_key,
                min_confirmations=self.config.taker_utxo_age,
                min_percent=self.config.taker_utxo_amtpercent,
                max_retries=self.config.taker_utxo_retries,
            )

            if not self._session.podle_commitment:
                reason = "Failed to generate PoDLE commitment"
                logger.error(reason)
                self._session.last_failure_reason = reason
                self.state = TakerState.FAILED
                return None

            max_replacement_attempts = self.config.max_maker_replacement_attempts
            if not await self._run_fill_with_replacements(
                destination=destination,
                selected_offers=selected_offers,
                required_features=required_features,
                mixdepth=mixdepth,
                get_private_key=get_private_key,
                max_replacement_attempts=max_replacement_attempts,
            ):
                return None

            if not await self._run_auth_with_replacements(
                required_features=required_features,
                max_replacement_attempts=max_replacement_attempts,
            ):
                return None

            # Phase 3: Build transaction
            self.state = TakerState.BUILDING_TX
            logger.info("Phase 3: Building transaction...")

            tx_success = await self._session._phase_build_tx(
                destination=destination,
                mixdepth=mixdepth,
            )
            if not tx_success:
                logger.error("Transaction build failed")
                self.state = TakerState.FAILED
                return None

            # Phase 4: Collect signatures
            self.state = TakerState.COLLECTING_SIGNATURES
            logger.info("Phase 4: Collecting signatures...")

            sig_success = await self._session._phase_collect_signatures()
            if not sig_success:
                logger.error("Signature collection failed")
                self.state = TakerState.FAILED
                return None

            return await self._finalize_and_broadcast(destination)

        except Exception as e:
            logger.error(f"CoinJoin failed: {e}")
            # Fire-and-forget notification for failed CoinJoin
            phase = self.state.value if hasattr(self, "state") else ""
            amount = self._session.cj_amount
            spawn_task(get_notifier().notify_coinjoin_failed(str(e), phase, amount))
            self.state = TakerState.FAILED
            return None
        finally:
            # Before !tx, failure is known not to have produced signatures and
            # locks can be released. At or after the signing boundary, renew
            # through the pending window because cancellation, confirmation
            # decline, or a failed broadcast can leave a usable transaction.
            if self.state != TakerState.COMPLETE:
                self.release_input_locks()

    async def _maybe_select_utxos_interactively(
        self, amount: int, mixdepth: int | None
    ) -> list[UTXOInfo] | None:
        """Run the interactive UTXO selector across the whole wallet.

        All mixdepths are displayed for a full-wallet overview. When
        ``mixdepth`` is ``None`` the user may select from any mixdepth (the
        TUI pins the source mixdepth to the first selected UTXO); when set,
        only that mixdepth is selectable and the rest is context.
        """
        if not self.config.select_utxos:
            logger.debug("Interactive UTXO selection not requested (--select-utxos not set)")
            return None

        from jmwallet.history import get_utxo_label
        from jmwallet.utxo_selector import select_utxos_interactive

        try:
            # Get ALL UTXOs (all mixdepths, including frozen/immature ones)
            # for display in the interactive selector. Ineligible UTXOs are
            # shown but rendered as unselectable ([-]) so the user sees the
            # full picture of their wallet.
            min_age = self.config.taker_utxo_age
            available_utxos: list[UTXOInfo] = []
            for md in range(self.wallet.mixdepth_count):
                available_utxos.extend(await self.wallet.get_utxos(md))
            if not available_utxos:
                reason = "No UTXOs in wallet"
                logger.error(reason)
                self._session.last_failure_reason = reason
                self.state = TakerState.FAILED
                return None

            # Check that at least some UTXOs are selectable (confirmed
            # enough, not frozen/locked, and in the pinned mixdepth if any).
            candidates = (
                available_utxos
                if mixdepth is None
                else [u for u in available_utxos if u.mixdepth == mixdepth]
            )
            locked_inputs = self.wallet.get_locked_input_outpoints()
            if not selectable_for_interactive(
                candidates, min_age, excluded_outpoints=locked_inputs
            ):
                where = "wallet" if mixdepth is None else f"mixdepth {mixdepth}"
                reason = (
                    f"No eligible UTXOs in {where} "
                    f"(all {len(candidates)} UTXOs are frozen, immature, locked, or in use)"
                )
                logger.error(reason)
                self._session.last_failure_reason = reason
                self.state = TakerState.FAILED
                return None

            # Populate labels for each UTXO based on history
            for utxo in available_utxos:
                if utxo.label is None:
                    utxo.label = get_utxo_label(
                        utxo.address,
                        self.config.data_dir,
                        wallet_fingerprint=self.wallet.wallet_fingerprint,
                    )

            logger.info(
                f"Launching interactive UTXO selector ({len(available_utxos)} available, "
                f"target amount: {amount} sats, sweep: {amount == 0})..."
            )
            manually_selected_utxos = select_utxos_interactive(
                available_utxos,
                amount,
                allowed_mixdepth=mixdepth,
                min_confirmations=min_age,
                excluded_outpoints=locked_inputs,
            )

            if not manually_selected_utxos:
                logger.info("UTXO selection cancelled by user")
                self.state = TakerState.CANCELLED
                return None

            total_selected = sum(u.value for u in manually_selected_utxos)
            logger.info(
                f"Manually selected {len(manually_selected_utxos)} UTXOs "
                f"(total: {total_selected:,} sats)"
            )

            # Validate selected UTXOs have sufficient funds (for non-sweep)
            if amount > 0 and total_selected < amount:
                logger.error(
                    f"Insufficient funds in selected UTXOs: "
                    f"have {total_selected:,} sats, need at least {amount:,} sats"
                )
                self.state = TakerState.FAILED
                return None
        except RuntimeError as e:
            logger.error(f"Interactive UTXO selection failed: {e}")
            self.state = TakerState.FAILED
            return None

        return manually_selected_utxos

    async def _run_auth_with_replacements(
        self, required_features: set[str] | None, max_replacement_attempts: int
    ) -> bool:
        self.state = TakerState.AUTHENTICATING
        logger.info("Phase 2: Sending !auth and receiving !ioauth...")

        auth_replacement_attempt = 0
        # Nicks that failed at any point during this auth stage (incompatible,
        # no/invalid !ioauth, failed replacement mini-fill). Hard-excluded from
        # re-selection so the same maker is never retried within this round.
        failed_nicks: set[str] = set()
        while True:
            auth_result = await self._session._phase_auth()

            if auth_result.success:
                return True

            for failed_nick in auth_result.failed_makers:
                self.orderbook_manager.add_ignored_maker(failed_nick)
                failed_nicks.add(failed_nick)
                logger.debug(f"Added {failed_nick} to ignored makers (failed auth)")

            if not auth_result.needs_replacement:
                logger.error("Auth phase failed")
                self.state = TakerState.FAILED
                return False

            # Top up the session back to minimum_makers. Replacement candidates
            # can themselves fail the mini-fill (offline maker, timeout), so
            # keep selecting substitutes until the session is whole again or
            # the replacement budget runs out.
            while auth_replacement_attempt < max_replacement_attempts:
                auth_replacement_attempt += 1
                needed = self.config.minimum_makers - len(self._session.maker_sessions)
                logger.info(
                    f"Attempting maker replacement in auth phase "
                    f"(attempt {auth_replacement_attempt}/{max_replacement_attempts}): "
                    f"need {needed} more makers"
                )

                current_session_nicks = set(self._session.maker_sessions.keys())
                replacement_offers, _ = self.orderbook_manager.select_makers(
                    cj_amount=self._session.cj_amount,
                    n=needed,
                    hard_exclude_nicks=current_session_nicks | failed_nicks,
                    required_features=required_features,
                )

                if len(replacement_offers) < needed:
                    logger.error(
                        f"Not enough replacement makers for auth phase: "
                        f"found {len(replacement_offers)}, need {needed}"
                    )
                    self.state = TakerState.FAILED
                    return False

                if not await self._fill_replacement_makers(replacement_offers, failed_nicks):
                    self.state = TakerState.FAILED
                    return False

                if len(self._session.maker_sessions) >= self.config.minimum_makers:
                    break
                logger.warning(
                    f"Still short of makers after replacement fill: "
                    f"{len(self._session.maker_sessions)} < {self.config.minimum_makers}"
                )
            else:
                logger.error(
                    f"Auth phase failed: could not assemble {self.config.minimum_makers} "
                    f"makers within {max_replacement_attempts} replacement attempts"
                )
                self.state = TakerState.FAILED
                return False

    async def _fill_replacement_makers(
        self, replacement_offers: dict[str, Any], failed_nicks: set[str]
    ) -> bool:
        """Run a mini fill phase for auth-stage replacement makers.

        Creates sessions, binds channels, sends !fill and processes the
        !pubkey responses via the shared helper (which also parses the
        features field, so replacement makers advertising neutrino_compat are
        not re-dropped in the next auth pass). Makers that do not produce a
        usable !pubkey are removed, ignored and added to ``failed_nicks`` so
        they are not re-selected within this round.

        Returns False only on unrecoverable state (missing PoDLE commitment or
        crypto session); "no maker responded" is left to the caller's
        replacement budget.
        """
        if not self._session.podle_commitment or not self._session.crypto_session:
            logger.error("Missing commitment or crypto session for replacement")
            return False

        for nick, offer in replacement_offers.items():
            self._session.maker_sessions[nick] = MakerSession(
                nick=nick, offer=offer, supports_neutrino_compat=False
            )
            logger.info(f"Added replacement maker for auth: {nick}")
        self._session.last_used_nicks.update(replacement_offers.keys())

        logger.info("Running fill phase for replacement makers...")
        new_maker_nicks = list(replacement_offers.keys())

        commitment_hex = self._session.podle_commitment.to_commitment_str()
        taker_pubkey = self._session.crypto_session.get_pubkey_hex()

        for nick in new_maker_nicks:
            binding = self.directory_client.bind_session(nick)
            session = self._session.maker_sessions[nick]
            if binding is None:
                logger.warning(f"No communication channel available for replacement maker {nick}")
                continue
            session.comm_channel = binding.channel_id
            if binding.is_direct:
                logger.debug(f"Will use DIRECT connection for replacement maker {nick}")
            else:
                logger.debug(f"Will use {binding.channel_id} for replacement maker {nick}")

        for nick in new_maker_nicks:
            session = self._session.maker_sessions[nick]
            fill_data = (
                f"{session.offer.oid} {self._session.cj_amount} {taker_pubkey} {commitment_hex}"
            )
            await self.directory_client.send_privmsg(
                nick,
                "fill",
                fill_data,
                log_routing=True,
                force_channel=session.comm_channel,
            )

        responses = await self.directory_client.wait_for_responses(
            expected_nicks=new_maker_nicks,
            expected_command="!pubkey",
            timeout=self.config.maker_timeout_sec,
        )

        for nick in new_maker_nicks:
            ready = False
            if nick in responses and not responses[nick].get("error"):
                try:
                    response_data = responses[nick]["data"].strip()
                    ready = self._session.process_pubkey_response(nick, response_data)
                    if ready:
                        logger.debug(f"Replacement maker {nick} ready")
                except Exception as e:
                    logger.warning(f"Failed to process {nick}: {e}")
            else:
                logger.warning(f"Replacement maker {nick} didn't respond to !fill")
            if not ready:
                self._session.maker_sessions.pop(nick, None)
                failed_nicks.add(nick)
                self.orderbook_manager.add_ignored_maker(nick)
        return True

    async def _run_fill_with_replacements(
        self,
        destination: str,
        selected_offers: dict[str, Any],
        required_features: set[str] | None,
        mixdepth: int,
        get_private_key: Any,
        max_replacement_attempts: int,
    ) -> bool:
        self.state = TakerState.FILLING
        logger.info("Phase 1: Sending !fill to makers...")
        directory_count = len(self.directory_client.clients)
        directories = [
            f"{client.host}:{client.port}" for client in self.directory_client.clients.values()
        ]
        logger.info(
            f"Routing via {directory_count} director{'y' if directory_count == 1 else 'ies'}: "
            f"{', '.join(directories)}"
        )
        if self.directory_client.prefer_direct_connections:
            logger.debug(
                "Direct connections preferred - will attempt to connect directly to makers"
            )
        else:
            logger.debug("Direct connections disabled - all messages relayed through directories")

        spawn_task(
            get_notifier().notify_coinjoin_start(
                self._session.cj_amount, len(self._session.maker_sessions), destination
            )
        )

        max_podle_retries = self.config.taker_utxo_retries
        replacement_attempt = 0
        for podle_retry in range(max_podle_retries):
            session_size_before_fill = len(self._session.maker_sessions)
            fill_result = await self._session._phase_fill()
            if fill_result.success:
                return True

            if fill_result.blacklist_makers and self._session.podle_commitment is not None:
                commitment_hex = self._session.podle_commitment.commitment.commitment.hex()
                try:
                    from jmcore.commitment_blacklist import add_commitment

                    add_commitment(commitment_hex)
                except Exception as exc:  # pragma: no cover - defensive
                    logger.warning(
                        f"Could not persist remotely-reported blacklisted commitment: {exc}"
                    )

            n_blacklisted = len(fill_result.blacklist_makers)
            majority_blacklist = (
                fill_result.blacklist_error
                and session_size_before_fill > 0
                and n_blacklisted * 2 >= session_size_before_fill
            )

            if fill_result.blacklist_error and not majority_blacklist:
                logger.warning(
                    f"Minority blacklist rejection from {fill_result.blacklist_makers} "
                    f"({n_blacklisted}/{session_size_before_fill}). Ignoring those makers "
                    "and trying replacement with the same commitment."
                )
                for failed_nick in fill_result.failed_makers:
                    self.orderbook_manager.add_ignored_maker(failed_nick)
                    logger.debug(f"Added {failed_nick} to ignored makers (minority blacklist)")
            elif fill_result.blacklist_error:
                logger.warning(
                    f"Majority blacklist rejection ({n_blacklisted}/{session_size_before_fill}) "
                    f"from {fill_result.blacklist_makers}. Rotating commitment and retrying."
                )
            elif fill_result.failed_makers:
                for failed_nick in fill_result.failed_makers:
                    self.orderbook_manager.add_ignored_maker(failed_nick)
                    logger.debug(f"Added {failed_nick} to ignored makers (failed fill)")

            if majority_blacklist:
                if podle_retry < max_podle_retries - 1:
                    logger.warning(
                        f"Commitment blacklisted, retrying with new NUMS index "
                        f"(attempt {podle_retry + 2}/{max_podle_retries})..."
                    )
                    new_commitment = self.podle_manager.generate_fresh_commitment(
                        wallet_utxos=self._session.preselected_utxos,
                        cj_amount=self._session.cj_amount,
                        private_key_getter=get_private_key,
                        min_confirmations=self.config.taker_utxo_age,
                        min_percent=self.config.taker_utxo_amtpercent,
                        max_retries=self.config.taker_utxo_retries,
                    )
                    if new_commitment is None:
                        added = self._session._expand_preselected_utxos_same_mixdepth(mixdepth)
                        if added > 0:
                            logger.info(
                                f"Preselected UTXOs exhausted for PoDLE; added {added} "
                                f"additional UTXO(s) from mixdepth {mixdepth}, which will "
                                "also be spent in the CoinJoin."
                            )
                            new_commitment = self.podle_manager.generate_fresh_commitment(
                                wallet_utxos=self._session.preselected_utxos,
                                cj_amount=self._session.cj_amount,
                                private_key_getter=get_private_key,
                                min_confirmations=self.config.taker_utxo_age,
                                min_percent=self.config.taker_utxo_amtpercent,
                                max_retries=self.config.taker_utxo_retries,
                            )
                    if new_commitment is None:
                        logger.error(
                            "No more PoDLE commitments available: all indices exhausted "
                            f"across all eligible UTXOs in mixdepth {mixdepth}"
                        )
                        self.state = TakerState.FAILED
                        return False

                    self._session.podle_commitment = new_commitment
                    self._session.maker_sessions = {
                        nick: MakerSession(nick=nick, offer=offer, supports_neutrino_compat=False)
                        for nick, offer in selected_offers.items()
                        if nick not in self.orderbook_manager.ignored_makers
                    }
                    continue

                logger.error(
                    f"Fill phase failed after {max_podle_retries} PoDLE commitment attempts"
                )
                self.state = TakerState.FAILED
                return False

            if fill_result.needs_replacement and replacement_attempt < max_replacement_attempts:
                replacement_attempt += 1
                needed = self.config.minimum_makers - len(self._session.maker_sessions)
                logger.info(
                    f"Attempting maker replacement (attempt {replacement_attempt}/"
                    f"{max_replacement_attempts}): need {needed} more makers"
                )

                current_session_nicks = set(self._session.maker_sessions.keys())
                hard_excludes = current_session_nicks | set(fill_result.failed_makers)
                replacement_offers, _ = self.orderbook_manager.select_makers(
                    cj_amount=self._session.cj_amount,
                    n=needed,
                    hard_exclude_nicks=hard_excludes,
                    required_features=required_features,
                )

                if len(replacement_offers) < needed:
                    logger.error(
                        "Not enough replacement makers available: "
                        f"found {len(replacement_offers)}, need {needed}"
                    )
                    self.state = TakerState.FAILED
                    return False

                for nick, offer in replacement_offers.items():
                    self._session.maker_sessions[nick] = MakerSession(
                        nick=nick, offer=offer, supports_neutrino_compat=False
                    )
                    logger.info(f"Added replacement maker: {nick}")
                selected_offers.update(replacement_offers)
                self._session.last_used_nicks.update(replacement_offers.keys())
                continue

            logger.error("Fill phase failed")
            self.state = TakerState.FAILED
            return False

        logger.error("Fill phase failed")
        self.state = TakerState.FAILED
        return False

    async def _finalize_and_broadcast(self, destination: str) -> str | None:
        # Final confirmation before broadcast
        num_taker_inputs = len(self._session.selected_utxos)
        num_maker_inputs = sum(len(s.utxos) for s in self._session.maker_sessions.values())
        total_inputs = num_taker_inputs + num_maker_inputs

        tx = deserialize_transaction(self._session.final_tx)
        total_outputs = len(tx.outputs)
        total_output_value = sum(out.value for out in tx.outputs)

        taker_input_value = sum(utxo.value for utxo in self._session.selected_utxos)
        maker_input_value = sum(
            utxo["value"]
            for session in self._session.maker_sessions.values()
            for utxo in session.utxos
        )
        total_input_value = taker_input_value + maker_input_value
        actual_mining_fee = total_input_value - total_output_value

        total_maker_fees = sum(
            calculate_cj_fee(session.offer, self._session.cj_amount)
            for session in self._session.maker_sessions.values()
        )
        total_cost = total_maker_fees + actual_mining_fee
        actual_vsize = calculate_tx_vsize(self._session.final_tx)
        actual_fee_rate = actual_mining_fee / actual_vsize if actual_vsize > 0 else 0.0

        logger.info("=" * 70)
        logger.info("FINAL TRANSACTION SUMMARY - Ready to broadcast")
        logger.info("=" * 70)
        logger.info(f"CoinJoin amount:      {self._session.cj_amount:,} sats")
        logger.info(f"Makers participating: {len(self._session.maker_sessions)}")
        logger.info(
            f"  Makers: {', '.join(nick[:10] + '...' for nick in self._session.maker_sessions)}"
        )
        logger.info(
            f"Transaction inputs:   {total_inputs} ({num_taker_inputs} yours, "
            f"{num_maker_inputs} makers)"
        )
        logger.info(f"Transaction outputs:  {total_outputs}")
        logger.info(f"Maker fees:           {total_maker_fees:,} sats")
        logger.info(
            f"Mining fee:           {actual_mining_fee:,} sats ({actual_fee_rate:.2f} sat/vB)"
        )
        logger.info(f"Total cost:           {total_cost:,} sats")
        logger.info(
            f"Transaction size:     {actual_vsize} vbytes ({len(self._session.final_tx)} bytes)"
        )
        logger.info("-" * 70)
        logger.info("Transaction hex (for manual verification/broadcast):")
        logger.info(self._session.final_tx.hex())
        logger.info("=" * 70)

        if hasattr(self, "confirmation_callback") and self.confirmation_callback:
            try:
                maker_details = []
                for nick, session in self._session.maker_sessions.items():
                    fee = calculate_cj_fee(session.offer, self._session.cj_amount)
                    bond_value = session.offer.fidelity_bond_value
                    location = None
                    for client in self.directory_client.clients.values():
                        location = client._active_peers.get(nick)
                        if location and location != "NOT-SERVING-ONION":
                            break
                    maker_details.append(
                        {
                            "nick": nick,
                            "fee": fee,
                            "bond_value": bond_value,
                            "location": location,
                        }
                    )

                confirmed = self.confirmation_callback(
                    maker_details=maker_details,
                    cj_amount=self._session.cj_amount,
                    total_fee=total_cost,
                    destination=destination,
                    mining_fee=actual_mining_fee,
                    fee_rate=actual_fee_rate,
                    stage="broadcast",
                )
                if not confirmed:
                    logger.warning("User declined final broadcast confirmation")
                    self._session._log_manual_csv_entry(
                        total_maker_fees, actual_mining_fee, destination
                    )
                    self.state = TakerState.FAILED
                    return None
            except Exception as e:
                logger.error(f"Final confirmation callback failed: {e}")
                self.state = TakerState.FAILED
                return None

        self.state = TakerState.BROADCASTING
        logger.info("Phase 5: Broadcasting transaction...")

        self._session.txid = await self._session._phase_broadcast()
        if not self._session.txid:
            logger.error("Broadcast failed")
            self.state = TakerState.FAILED
            return None

        self.state = TakerState.COMPLETE
        logger.info(f"CoinJoin COMPLETE! txid: {self._session.txid}")

        try:
            updated = update_taker_awaiting_transaction_broadcast(
                destination_address=self._session.cj_destination,
                change_address=self._session.taker_change_address,  # Empty string if no change
                txid=self._session.txid,
                mining_fee=actual_mining_fee,
                data_dir=self.config.data_dir,
                wallet_fingerprint=self.wallet.wallet_fingerprint,
            )
            if updated:
                logger.debug(
                    f"Updated history entry for CJ txid {self._session.txid[:16]}..., "
                    f"mining_fee={actual_mining_fee} sats"
                )
            else:
                logger.warning(
                    f"No matching 'Awaiting transaction' entry found for "
                    f"{self._session.cj_destination[:20]}... - history may be inconsistent"
                )

            destination_vout = self._session._get_taker_cj_output_index()
            await self._update_pending_transaction_now(
                self._session.txid,
                self._session.cj_destination,
                destination_vout if destination_vout is not None else -1,
                len(self._session.maker_sessions),
            )
        except Exception as e:
            logger.warning(f"Failed to update CoinJoin history: {e}")

        total_fees = total_maker_fees + actual_mining_fee
        spawn_task(
            get_notifier().notify_coinjoin_complete(
                self._session.txid,
                self._session.cj_amount,
                len(self._session.maker_sessions),
                total_fees,
            )
        )

        return self._session.txid

    async def run_schedule(self, schedule: Schedule) -> bool:
        """
        Run a tumbler-style schedule of CoinJoins.

        Args:
            schedule: Schedule with multiple CoinJoin entries

        Returns:
            True if all entries completed successfully
        """
        self.schedule = schedule

        while not schedule.is_complete():
            entry = schedule.current_entry()
            if not entry:
                break

            logger.info(
                f"Running schedule entry {schedule.current_index + 1}/{len(schedule.entries)}"
            )

            # Calculate actual amount
            if entry.amount_fraction is not None:
                # Fraction of balance
                balance = await self.wallet.get_balance(entry.mixdepth)
                amount = int(balance * entry.amount_fraction)
            else:
                assert entry.amount is not None
                amount = entry.amount

            # Execute CoinJoin
            txid = await self.do_coinjoin(
                amount=amount,
                destination=entry.destination,
                mixdepth=entry.mixdepth,
                counterparty_count=entry.counterparty_count,
            )

            if not txid:
                logger.error(f"Schedule entry {schedule.current_index + 1} failed")
                return False

            # Advance schedule
            schedule.advance()

            # Wait between CoinJoins
            if entry.wait_time > 0 and not schedule.is_complete():
                logger.info(f"Waiting {entry.wait_time}s before next CoinJoin...")
                await asyncio.sleep(entry.wait_time)

        logger.info("Schedule complete!")
        return True

Attributes

backend = backend instance-attribute

config = config instance-attribute

confirmation_callback = confirmation_callback instance-attribute

directory_client = MultiDirectoryClient(directory_servers=(config.directory_servers), network=(config.network.value), nick_identity=(self.nick_identity), socks_host=(config.socks_host), socks_port=(config.socks_port), connection_timeout=(config.connection_timeout), nick_auth_mode=(config.nick_auth_mode), nick_auth_directory_ids=(config.nick_auth_directory_ids), neutrino_compat=neutrino_compat, stream_isolation=(config.stream_isolation)) instance-attribute

last_failure_reason: str | None property

Reason the most recent do_coinjoin call failed (or None).

Forwarded from the per-round :class:CoinJoinSession so external consumers (e.g. the tumbler runner) can surface why a round did not broadcast without reaching into private session state.

last_source_mixdepth: int | None = None instance-attribute

last_used_nicks: set[str] property

Maker nicks used by the most recent do_coinjoin call.

nick = self.nick_identity.nick instance-attribute

nick_identity = NickIdentity(JM_VERSION) instance-attribute

orderbook_manager = OrderbookManager(config.max_cj_fee, bondless_makers_allowance=(config.bondless_makers_allowance), bondless_require_zero_fee=(config.bondless_makers_allowance_require_zero_fee), data_dir=(config.data_dir), own_wallet_nicks=own_wallet_nicks) instance-attribute

podle_manager = PoDLEManager(config.data_dir) instance-attribute

running = False instance-attribute

schedule: Schedule | None = None instance-attribute

state = TakerState.IDLE instance-attribute

wallet = wallet instance-attribute

Methods:

__init__(wallet: WalletService, backend: BlockchainBackend, config: TakerConfig, confirmation_callback: Any | None = None)

Initialize the Taker.

Args: wallet: Wallet service for UTXO management and signing backend: Blockchain backend for broadcasting config: Taker configuration confirmation_callback: Optional callback for user confirmation before proceeding

Source code in taker/src/taker/taker.py

def __init__(
    self,
    wallet: WalletService,
    backend: BlockchainBackend,
    config: TakerConfig,
    confirmation_callback: Any | None = None,
):
    """
    Initialize the Taker.

    Args:
        wallet: Wallet service for UTXO management and signing
        backend: Blockchain backend for broadcasting
        config: Taker configuration
        confirmation_callback: Optional callback for user confirmation before proceeding
    """
    self.wallet = wallet
    self.backend = backend
    self.config = config
    self.confirmation_callback = confirmation_callback

    self.nick_identity = NickIdentity(JM_VERSION)
    self.nick = self.nick_identity.nick
    self.state = TakerState.IDLE
    # Source mixdepth of the most recent ``do_coinjoin`` call. With
    # interactive selection this is derived from the selected UTXOs, so
    # callers (e.g. the CLI confirmation prompt) can display it.
    self.last_source_mixdepth: int | None = None

    # Advertise neutrino_compat if our backend can provide extended UTXO metadata.
    # This tells other peers that we can provide scriptpubkey and blockheight.
    # Full nodes (Bitcoin Core) can provide this; light clients (Neutrino) cannot.
    neutrino_compat = backend.can_provide_neutrino_metadata()

    # Directory client
    self.directory_client = MultiDirectoryClient(
        directory_servers=config.directory_servers,
        network=config.network.value,
        nick_identity=self.nick_identity,
        socks_host=config.socks_host,
        socks_port=config.socks_port,
        connection_timeout=config.connection_timeout,
        nick_auth_mode=config.nick_auth_mode,
        nick_auth_directory_ids=config.nick_auth_directory_ids,
        neutrino_compat=neutrino_compat,
        stream_isolation=config.stream_isolation,
    )

    # Orderbook manager
    # Read maker nick from state file to exclude from peer selection (self-CoinJoin protection)
    own_wallet_nicks: set[str] = set()
    maker_nick = read_nick_state(config.data_dir, "maker")
    if maker_nick:
        own_wallet_nicks.add(maker_nick)
        logger.info(f"Self-CoinJoin protection: excluding maker nick {maker_nick}")

    self.orderbook_manager = OrderbookManager(
        config.max_cj_fee,
        bondless_makers_allowance=config.bondless_makers_allowance,
        bondless_require_zero_fee=config.bondless_makers_allowance_require_zero_fee,
        data_dir=config.data_dir,
        own_wallet_nicks=own_wallet_nicks,
    )

    # PoDLE manager for commitment tracking
    self.podle_manager = PoDLEManager(config.data_dir)

    # Per-CoinJoin state and protocol phases live in a dedicated
    # ``CoinJoinSession``. ``Taker`` owns persistent infrastructure
    # (wallet, backend, config, directory client, orderbook manager,
    # PoDLE manager, schedule) and delegates the protocol phases to the
    # session, which is reset at the start of each ``do_coinjoin`` call.
    self._session = CoinJoinSession()
    self._session.attach(self)
    self._round_lock = asyncio.Lock()

    # Schedule for tumbler-style operations
    self.schedule: Schedule | None = None

    # Background task tracking
    self.running = False
    self._background_tasks: list[asyncio.Task[None]] = []

check_utxo_eligibility(amount: int, mixdepth: int | None) -> str | None async

Validate that mixdepth can fund a CoinJoin of amount.

Runs the same eligibility filters used later in :meth:do_coinjoin (confirmations, frozen, fidelity bonds, in-flight locks, the mixdepth-0 merge restriction and the PoDLE size requirement) before any network operation, so an ineligible wallet fails fast instead of after a long directory/orderbook/bond cycle (issue #528).

Args: amount: Target amount in satoshis (0 for sweep). mixdepth: Source mixdepth. None is only meaningful with interactive selection (select_utxos), where it means "any mixdepth" (the source is derived from the selection later); otherwise it falls back to mixdepth 0.

Returns: None when a CoinJoin can proceed, otherwise a human-readable reason describing why it cannot.

Source code in taker/src/taker/taker.py

async def check_utxo_eligibility(self, amount: int, mixdepth: int | None) -> str | None:
    """Validate that ``mixdepth`` can fund a CoinJoin of ``amount``.

    Runs the same eligibility filters used later in :meth:`do_coinjoin`
    (confirmations, frozen, fidelity bonds, in-flight locks, the mixdepth-0
    merge restriction and the PoDLE size requirement) *before* any network
    operation, so an ineligible wallet fails fast instead of after a long
    directory/orderbook/bond cycle (issue #528).

    Args:
        amount: Target amount in satoshis (``0`` for sweep).
        mixdepth: Source mixdepth. ``None`` is only meaningful with
            interactive selection (``select_utxos``), where it means "any
            mixdepth" (the source is derived from the selection later);
            otherwise it falls back to mixdepth 0.

    Returns:
        ``None`` when a CoinJoin can proceed, otherwise a human-readable
        reason describing why it cannot.
    """
    min_conf = self.config.taker_utxo_age

    # Interactive selection follows different rules (the user may pick
    # unlocked fidelity bonds and is not bound to auto-selection), so only
    # require that *something* is selectable here.
    if self.config.select_utxos:
        mixdepths = (
            [mixdepth] if mixdepth is not None else list(range(self.wallet.mixdepth_count))
        )
        utxos = []
        for md in mixdepths:
            utxos.extend(await self.wallet.get_utxos(md))
        reserved = self.wallet.get_locked_input_outpoints()
        if not selectable_for_interactive(utxos, min_conf, excluded_outpoints=reserved):
            if mixdepth is not None:
                return classify_utxos(
                    utxos, mixdepth, min_conf, reserved_outpoints=reserved
                ).no_eligible_reason()
            return (
                "No selectable UTXOs in any mixdepth (all UTXOs are "
                "frozen, immature, locked fidelity bonds, or in use)"
            )
        return None

    if mixdepth is None:
        mixdepth = 0
    utxos = await self.wallet.get_utxos(mixdepth)

    reserved = self.wallet.get_locked_input_outpoints()
    breakdown = classify_utxos(utxos, mixdepth, min_conf, reserved_outpoints=reserved)

    if not breakdown.eligible:
        return breakdown.no_eligible_reason()

    # Sweep spends every eligible UTXO; a non-empty pool is sufficient.
    is_sweep = amount == 0
    if is_sweep:
        return None

    # PoDLE necessary condition: a commitment needs a UTXO worth at least
    # ``taker_utxo_amtpercent`` of the amount. Without one the round always
    # fails at commitment generation, so reject early with a clear message.
    if not podle_threshold_met(
        breakdown.eligible, amount, min_conf, self.config.taker_utxo_amtpercent
    ):
        min_value = int(amount * self.config.taker_utxo_amtpercent / 100)
        return (
            f"No eligible UTXO in mixdepth {mixdepth} is large enough for the "
            f"PoDLE commitment: need at least {min_value:,} sats "
            f"({self.config.taker_utxo_amtpercent}% of {amount:,} sats, "
            f"taker_utxo_amtpercent). Use a larger UTXO or lower the amount."
        )

    # Amount coverage: dry-run the exact selection used later so the verdict
    # matches reality (including the mixdepth-0 merge restriction).
    try:
        self.wallet.select_utxos(
            mixdepth,
            amount,
            min_conf,
            exclude=reserved,
        )
    except ValueError as exc:
        return _append_confirmation_hint(str(exc), min_conf)

    return None

connect() -> None async

Connect to directory servers and start background tasks.

This should be called after sync_wallet() and any fund validation.

Source code in taker/src/taker/taker.py

async def connect(self) -> None:
    """
    Connect to directory servers and start background tasks.

    This should be called after sync_wallet() and any fund validation.
    """
    # Connect to directory servers
    logger.info("Connecting to directory servers...")
    connected = await self.directory_client.connect_all()

    if connected == 0:
        raise RuntimeError("Failed to connect to any directory server")

    logger.info(f"Connected to {connected} directory servers")

    # Mark as running and start background tasks
    self.running = True

    # Start pending transaction monitor
    monitor_task = asyncio.create_task(self._monitor_pending_transactions())
    self._background_tasks.append(monitor_task)

    # Start periodic rescan task (useful for schedule mode)
    rescan_task = asyncio.create_task(self._periodic_rescan())
    self._background_tasks.append(rescan_task)

    # Start periodic directory connection status logging task
    conn_status_task = asyncio.create_task(self._periodic_directory_connection_status())
    self._background_tasks.append(conn_status_task)

do_coinjoin(amount: int, destination: str, mixdepth: int | None = None, counterparty_count: int | None = None, exclude_nicks: set[str] | None = None) -> str | None async

Run one CoinJoin with fresh, non-reusable per-round state.

Source code in taker/src/taker/taker.py

async def do_coinjoin(
    self,
    amount: int,
    destination: str,
    mixdepth: int | None = None,
    counterparty_count: int | None = None,
    exclude_nicks: set[str] | None = None,
) -> str | None:
    """Run one CoinJoin with fresh, non-reusable per-round state."""
    if self._round_lock.locked():
        logger.error("A CoinJoin round is already active on this Taker instance")
        return None

    async with self._round_lock:
        session = CoinJoinSession()
        session.attach(self)
        self._session = session
        self.state = TakerState.IDLE
        return await self._do_coinjoin(
            amount=amount,
            destination=destination,
            mixdepth=mixdepth,
            counterparty_count=counterparty_count,
            exclude_nicks=exclude_nicks,
        )

release_input_locks() -> None

Clean up persisted CoinJoin locks held on this round's taker inputs.

Pre-sign failures release immediately. Once signatures may exist, the owner-qualified leases are renewed through the pending window instead.

Source code in taker/src/taker/taker.py

def release_input_locks(self) -> None:
    """Clean up persisted CoinJoin locks held on this round's taker inputs.

    Pre-sign failures release immediately. Once signatures may exist, the
    owner-qualified leases are renewed through the pending window instead.
    """
    if self._session and self._session.signing_boundary_crossed:
        self._session.retain_input_locks()
        return
    if self._session and self._session.reserved_inputs:
        try:
            self.wallet.release_coinjoin_inputs(
                self._session.reserved_inputs,
                owner=self._session.input_lock_owner,
            )
        except Exception as e:  # pragma: no cover - best-effort cleanup
            logger.debug(f"Failed to release taker input locks: {e}")
        self._session.reserved_inputs = set()

run_schedule(schedule: Schedule) -> bool async

Run a tumbler-style schedule of CoinJoins.

Args: schedule: Schedule with multiple CoinJoin entries

Returns: True if all entries completed successfully

Source code in taker/src/taker/taker.py

async def run_schedule(self, schedule: Schedule) -> bool:
    """
    Run a tumbler-style schedule of CoinJoins.

    Args:
        schedule: Schedule with multiple CoinJoin entries

    Returns:
        True if all entries completed successfully
    """
    self.schedule = schedule

    while not schedule.is_complete():
        entry = schedule.current_entry()
        if not entry:
            break

        logger.info(
            f"Running schedule entry {schedule.current_index + 1}/{len(schedule.entries)}"
        )

        # Calculate actual amount
        if entry.amount_fraction is not None:
            # Fraction of balance
            balance = await self.wallet.get_balance(entry.mixdepth)
            amount = int(balance * entry.amount_fraction)
        else:
            assert entry.amount is not None
            amount = entry.amount

        # Execute CoinJoin
        txid = await self.do_coinjoin(
            amount=amount,
            destination=entry.destination,
            mixdepth=entry.mixdepth,
            counterparty_count=entry.counterparty_count,
        )

        if not txid:
            logger.error(f"Schedule entry {schedule.current_index + 1} failed")
            return False

        # Advance schedule
        schedule.advance()

        # Wait between CoinJoins
        if entry.wait_time > 0 and not schedule.is_complete():
            logger.info(f"Waiting {entry.wait_time}s before next CoinJoin...")
            await asyncio.sleep(entry.wait_time)

    logger.info("Schedule complete!")
    return True

start() -> None async

Start the taker: sync wallet and connect to directory servers.

This is a convenience method that calls sync_wallet() followed by connect(). For early fund validation, call sync_wallet() first, validate, then call connect().

Source code in taker/src/taker/taker.py

async def start(self) -> None:
    """
    Start the taker: sync wallet and connect to directory servers.

    This is a convenience method that calls sync_wallet() followed by connect().
    For early fund validation, call sync_wallet() first, validate, then call connect().
    """
    await self.sync_wallet()
    await self.connect()

stop(*, close_wallet: bool = True) -> None async

Stop the taker and close connections.

Args: close_wallet: If True (the default), also close the wallet's backend connection. Pass False when the wallet is shared with another component (e.g. a jmwalletd tumbler runner that will reuse the same :class:~jmwallet.wallet.service.WalletService instance across multiple taker phases) to avoid tearing down a still-in-use wallet.

Source code in taker/src/taker/taker.py

async def stop(self, *, close_wallet: bool = True) -> None:
    """Stop the taker and close connections.

    Args:
        close_wallet: If ``True`` (the default), also close the wallet's
            backend connection. Pass ``False`` when the wallet is shared
            with another component (e.g. a jmwalletd tumbler runner that
            will reuse the same :class:`~jmwallet.wallet.service.WalletService`
            instance across multiple taker phases) to avoid tearing down a
            still-in-use wallet.
    """
    logger.info("Stopping taker...")
    self.running = False

    # Cancel all background tasks
    for task in self._background_tasks:
        task.cancel()

    if self._background_tasks:
        await asyncio.gather(*self._background_tasks, return_exceptions=True)
    self._background_tasks.clear()

    await self.directory_client.close_all()
    if close_wallet:
        await self.wallet.close()
    logger.info("Taker stopped")

sync_wallet() -> int async

Sync the wallet and return total balance.

This method is separated from start() to allow callers to check funds before connecting to directory servers (avoiding unnecessary network connections when funds are insufficient).

Returns: Total wallet balance in satoshis.

Source code in taker/src/taker/taker.py

async def sync_wallet(self) -> int:
    """
    Sync the wallet and return total balance.

    This method is separated from start() to allow callers to check
    funds before connecting to directory servers (avoiding unnecessary
    network connections when funds are insufficient).

    Returns:
        Total wallet balance in satoshis.
    """
    logger.info(f"Starting taker (nick: {self.nick})")

    # Log wallet name if using descriptor wallet backend
    from jmwallet.backends.descriptor_wallet import DescriptorWalletBackend

    if isinstance(self.backend, DescriptorWalletBackend):
        logger.info(f"Using wallet: {self.backend.wallet_name}")

    # Initialize commitment blacklist with configured data directory
    set_blacklist_path(data_dir=self.config.data_dir)

    # Sync wallet
    logger.info("Syncing wallet...")

    # Setup descriptor wallet if needed (one-time operation)
    if isinstance(self.backend, DescriptorWalletBackend):
        if not await self.wallet.is_descriptor_wallet_ready():
            logger.info("Descriptor wallet not set up. Importing descriptors...")
            await self.wallet.setup_descriptor_wallet(rescan=True)
            logger.info("Descriptor wallet setup complete")

        # Use fast descriptor wallet sync
        await self.wallet.sync_with_descriptor_wallet()
    else:
        # Use standard sync (BIP157/158 for neutrino, mempool API, etc.)
        await self.wallet.sync_all()
    await self.wallet.reconstruct_imported_state_safe()

    total_balance = await self.wallet.get_total_balance()
    logger.info(f"Wallet synced. Total balance: {total_balance:,} sats")

    return total_balance

TakerState

Bases: StrEnum

Taker protocol states.

Source code in taker/src/taker/models.py

class TakerState(StrEnum):
    """Taker protocol states."""

    IDLE = "idle"
    FETCHING_ORDERBOOK = "fetching_orderbook"
    SELECTING_MAKERS = "selecting_makers"
    FILLING = "filling"
    AUTHENTICATING = "authenticating"
    BUILDING_TX = "building_tx"
    COLLECTING_SIGNATURES = "collecting_signatures"
    BROADCASTING = "broadcasting"
    COMPLETE = "complete"
    FAILED = "failed"
    CANCELLED = "cancelled"  # User cancelled the operation

Attributes

AUTHENTICATING = 'authenticating' class-attribute instance-attribute

BROADCASTING = 'broadcasting' class-attribute instance-attribute

BUILDING_TX = 'building_tx' class-attribute instance-attribute

CANCELLED = 'cancelled' class-attribute instance-attribute

COLLECTING_SIGNATURES = 'collecting_signatures' class-attribute instance-attribute

COMPLETE = 'complete' class-attribute instance-attribute

FAILED = 'failed' class-attribute instance-attribute

FETCHING_ORDERBOOK = 'fetching_orderbook' class-attribute instance-attribute

FILLING = 'filling' class-attribute instance-attribute

IDLE = 'idle' class-attribute instance-attribute

SELECTING_MAKERS = 'selecting_makers' class-attribute instance-attribute

Functions:

warn_if_destination_script_mismatch(destination: str) -> str | None

Emit a warning when the destination address does not match the wallet's native script type. Returns the detected destination type on mismatch, None otherwise (matched type, or unparseable address - the canonical validation error is produced later in the pipeline).

See issue #113.

Source code in taker/src/taker/taker.py

def warn_if_destination_script_mismatch(destination: str) -> str | None:
    """
    Emit a warning when the destination address does not match the wallet's
    native script type. Returns the detected destination type on mismatch,
    None otherwise (matched type, or unparseable address - the canonical
    validation error is produced later in the pipeline).

    See issue #113.
    """
    try:
        dest_type = get_address_type(destination)
    except ValueError:
        return None
    if dest_type == _WALLET_OUTPUT_SCRIPT_TYPE:
        return None
    logger.warning(
        f"Destination address {destination} is {dest_type} but wallet is "
        f"{_WALLET_OUTPUT_SCRIPT_TYPE} (native segwit). Mixing script types "
        "in CoinJoin outputs fingerprints your output and reduces the "
        "effective anonymity set. Consider sending to a bech32 "
        "(bc1q.../tb1q...) address instead."
    )
    return dest_type