Nonce Management for AI Agents: Avoiding Transaction Failures

# DANGEROUS: Both coroutines read nonce=43 at the same time
# because await is a yield point between read and use

async def bad_send_tx(web3, account, to, value):
    nonce = await web3.eth.get_transaction_count(account, 'pending')
    # ← YIELD POINT: another coroutine can run here and also read nonce=43
    tx = {'nonce': nonce, 'to': to, 'value': value, ...}
    signed = account.sign_transaction(tx)
    # ← YIELD POINT: again vulnerable
    return await web3.eth.send_raw_transaction(signed.rawTransaction)

# If two coroutines run concurrently, both get nonce=43:
await asyncio.gather(
    bad_send_tx(w3, acct, addr1, value1),  # gets nonce 43
    bad_send_tx(w3, acct, addr2, value2),  # ALSO gets nonce 43 — COLLISION
)

"""
Production-grade NonceManager for AI agents.
Thread-safe, async-safe, with gap detection and recovery.
Compatible with Purple Flea Wallet API.
"""

import asyncio
import logging
import time
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Dict, Optional, Set
import aiohttp

log = logging.getLogger('nonce_manager')

# Chains where nonce management applies (non-EVM chains handled differently)
EVM_CHAINS = {"ETH", "MATIC", "BNB", "AVAX"}
STUCK_TX_TIMEOUT_S = 300    # 5 minutes before a tx is considered stuck
GAS_BUMP_FACTOR   = 1.15   # 15% gas bump for replacement transactions
MAX_PENDING_QUEUE = 50     # max unconfirmed txs before we pause submission


@dataclass
class PendingTx:
    tx_hash: str
    nonce: int
    gas_price_gwei: float
    submitted_at: float
    chain: str
    description: str = ""


class NonceManager:
    """
    Thread-safe nonce manager for concurrent EVM transaction submission.

    Features:
    - Atomic nonce allocation via asyncio.Lock
    - Pending transaction tracking per chain
    - Gap detection with automatic fill-gap recovery
    - Stuck transaction detection and gas bumping
    - Mempool reconciliation via RPC
    """

    def __init__(self, rpc_urls: Dict[str, str], address: str):
        # RPC endpoints per chain
        self.rpc_urls  = rpc_urls
        self.address   = address.lower()

        # Nonce state: chain → current local nonce
        self._local_nonce: Dict[str, int] = {}

        # One lock per chain — prevents concurrent nonce reads on same chain
        self._locks: Dict[str, asyncio.Lock] = defaultdict(asyncio.Lock)

        # Track pending transactions: chain → {nonce: PendingTx}
        self._pending: Dict[str, Dict[int, PendingTx]] = defaultdict(dict)

        # Set of nonces known to be confirmed: chain → set of nonces
        self._confirmed: Dict[str, Set[int]] = defaultdict(set)

        self._session: Optional[aiohttp.ClientSession] = None

    async def __aenter__(self):
        self._session = aiohttp.ClientSession()
        return self

    async def __aexit__(self, *args):
        if self._session:
            await self._session.close()

    async def _rpc(self, chain: str, method: str, params: list) -> dict:
        """Execute JSON-RPC call to chain node."""
        payload = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": method,
            "params": params,
        }
        async with self._session.post(
            self.rpc_urls[chain], json=payload
        ) as resp:
            data = await resp.json()
            if "error" in data:
                raise RuntimeError(f"RPC error on {chain}: {data['error']}")
            return data["result"]

    async def _fetch_pending_nonce(self, chain: str) -> int:
        """Get pending nonce from chain node."""
        result = await self._rpc(
            chain, "eth_getTransactionCount", [self.address, "pending"]
        )
        return int(result, 16)

    async def _fetch_confirmed_nonce(self, chain: str) -> int:
        """Get confirmed (latest) nonce from chain node."""
        result = await self._rpc(
            chain, "eth_getTransactionCount", [self.address, "latest"]
        )
        return int(result, 16)

    async def acquire_nonce(self, chain: str) -> int:
        """
        Atomically acquire the next nonce for a chain.

        This is the core method — always use this instead of
        querying the RPC directly. Guarantees no two callers
        get the same nonce even under concurrent load.
        """
        if chain not in EVM_CHAINS:
            raise ValueError(f"{chain} does not use EVM nonces")

        async with self._locks[chain]:
            # Initialize local nonce from chain if not yet cached
            if chain not in self._local_nonce:
                self._local_nonce[chain] = await self._fetch_pending_nonce(chain)
                log.info(f"[{chain}] Initialized local nonce: {self._local_nonce[chain]}")

            # Check if we are overloading the mempool
            pending_count = len(self._pending[chain])
            if pending_count >= MAX_PENDING_QUEUE:
                log.warning(f"[{chain}] Pending queue full ({pending_count}), pausing")
                await self._wait_for_confirmation(chain)

            nonce = self._local_nonce[chain]
            self._local_nonce[chain] += 1
            log.debug(f"[{chain}] Allocated nonce {nonce}")
            return nonce

    def register_pending(
        self,
        chain: str,
        nonce: int,
        tx_hash: str,
        gas_price_gwei: float,
        description: str = "",
    ):
        """Register a submitted transaction as pending."""
        self._pending[chain][nonce] = PendingTx(
            tx_hash=tx_hash,
            nonce=nonce,
            gas_price_gwei=gas_price_gwei,
            submitted_at=time.time(),
            chain=chain,
            description=description,
        )
        log.debug(f"[{chain}] Registered pending nonce={nonce} hash={tx_hash[:10]}...")

    async def reconcile(self, chain: str):
        """
        Reconcile local nonce state with on-chain state.
        Call this periodically (e.g., every 60s) or after
        detecting anomalies.
        """
        async with self._locks[chain]:
            on_chain_confirmed = await self._fetch_confirmed_nonce(chain)
            on_chain_pending   = await self._fetch_pending_nonce(chain)

            # Remove confirmed txs from pending dict
            confirmed_nonces = [n for n in self._pending[chain] if n < on_chain_confirmed]
            for n in confirmed_nonces:
                log.info(f"[{chain}] Nonce {n} confirmed")
                self._confirmed[chain].add(n)
                del self._pending[chain][n]

            # Detect gap: confirmed < on_chain_pending means the node sees
            # pending txs we might not have. Update local nonce to be safe.
            if on_chain_pending > self._local_nonce.get(chain, 0):
                log.warning(
                    f"[{chain}] Local nonce {self._local_nonce.get(chain)} behind"
                    f" pending nonce {on_chain_pending} — syncing"
                )
                self._local_nonce[chain] = on_chain_pending

            log.info(
                f"[{chain}] Reconciled: confirmed={on_chain_confirmed}"
                f" pending_queue={on_chain_pending}"
                f" local={self._local_nonce.get(chain, 'uninit')}"
            )

    async def detect_stuck(self, chain: str) -> list[PendingTx]:
        """Return list of transactions stuck beyond STUCK_TX_TIMEOUT_S."""
        now = time.time()
        stuck = [
            tx for tx in self._pending[chain].values()
            if now - tx.submitted_at > STUCK_TX_TIMEOUT_S
        ]
        if stuck:
            log.warning(f"[{chain}] {len(stuck)} stuck transactions detected")
        return stuck

    async def fill_gap(
        self,
        chain: str,
        gap_nonce: int,
        signer_fn,
        current_gas_gwei: float,
    ) -> str:
        """
        Fill a nonce gap by submitting a 0-value self-transfer.
        signer_fn(tx_dict) → signed transaction bytes
        """
        log.warning(f"[{chain}] Filling gap at nonce {gap_nonce}")
        cancel_tx = {
            "nonce": gap_nonce,
            "to": self.address,
            "value": 0,
            "gas": 21000,
            "maxFeePerGas": int(current_gas_gwei * 1e9 * 1.5),
            "maxPriorityFeePerGas": int(2e9),
        }
        signed_bytes = signer_fn(cancel_tx)
        result = await self._rpc(
            chain, "eth_sendRawTransaction", ["0x" + signed_bytes.hex()]
        )
        log.info(f"[{chain}] Gap fill tx: {result}")
        self.register_pending(chain, gap_nonce, result, current_gas_gwei * 1.5, "gap_fill")
        return result

    async def _wait_for_confirmation(self, chain: str, timeout_s: float = 120):
        """Block until pending queue drops below half of MAX_PENDING_QUEUE."""
        deadline = time.time() + timeout_s
        while len(self._pending[chain]) >= MAX_PENDING_QUEUE // 2:
            if time.time() > deadline:
                raise TimeoutError(f"[{chain}] Pending queue still full after {timeout_s}s")
            await asyncio.sleep(5)
            await self.reconcile(chain)

    def stats(self) -> Dict:
        return {
            chain: {
                "local_nonce": self._local_nonce.get(chain, "uninit"),
                "pending_count": len(self._pending[chain]),
                "confirmed_count": len(self._confirmed[chain]),
            }
            for chain in EVM_CHAINS
        }

import asyncio
from nonce_manager import NonceManager

RPC_URLS = {
    "ETH":   "https://eth-mainnet.g.alchemy.com/v2/YOUR_KEY",
    "MATIC": "https://polygon-rpc.com",
    "BNB":   "https://bsc-dataseed.binance.org",
    "AVAX":  "https://api.avax.network/ext/bc/C/rpc",
}

AGENT_ADDRESS = "0xYourAgentAddress"


async def place_bet(
    nm: NonceManager,
    chain: str,
    game: str,
    amount: float,
    signer,
) -> str:
    """Place a casino bet with nonce-safe transaction submission."""
    # ✅ Acquire nonce BEFORE any async work (atomic allocation)
    nonce = await nm.acquire_nonce(chain)

    try:
        # Build casino API transaction payload
        tx = {
            "nonce": nonce,
            "to": "0xPurpleFleCasinoContract",
            "data": encode_bet(game, amount),
            "gas": 200_000,
            "maxFeePerGas": int(30e9),
            "maxPriorityFeePerGas": int(2e9),
        }

        signed = signer(tx)
        tx_hash = await nm._rpc(chain, "eth_sendRawTransaction", [signed.hex()])
        nm.register_pending(chain, nonce, tx_hash, 30.0, f"casino:{game}:${amount}")

        log.info(f"Bet placed: {game} ${amount} nonce={nonce} tx={tx_hash[:10]}...")
        return tx_hash

    except Exception as e:
        # If submission failed, we have a nonce gap — fill it immediately
        log.error(f"Bet submission failed at nonce {nonce}: {e}")
        await nm.fill_gap(chain, nonce, signer, current_gas_gwei=35.0)
        raise


async def run_multi_game_agent():
    async with NonceManager(RPC_URLS, AGENT_ADDRESS) as nm:
        signer = get_signer()  # Your signing function

        # Start background reconciliation task
        asyncio.create_task(reconcile_loop(nm))

        # Place 5 concurrent bets safely
        results = await asyncio.gather(*[
            place_bet(nm, "ETH", "blackjack", 50,  signer),
            place_bet(nm, "ETH", "roulette", 25,  signer),
            place_bet(nm, "ETH", "dice",      10,  signer),
            place_bet(nm, "ETH", "slots",     5,   signer),
            place_bet(nm, "ETH", "poker",     100, signer),
        ])
        # ✅ All 5 bets get sequential nonces 43,44,45,46,47 — no collisions
        log.info(f"All bets placed: {results}")
        log.info(f"Nonce stats: {nm.stats()}")


async def reconcile_loop(nm: NonceManager):
    while True:
        await asyncio.sleep(60)
        for chain in EVM_CHAINS:
            try:
                await nm.reconcile(chain)
                stuck = await nm.detect_stuck(chain)
                if stuck:
                    log.warning(f"Stuck txs on {chain}: {[t.nonce for t in stuck]}")
            except Exception as e:
                log.error(f"Reconcile failed for {chain}: {e}")

import aiohttp

async def send_via_purpleflea(
    api_key: str,
    chain: str,
    to: str,
    value_wei: int,
    data: str = "",
) -> dict:
    """
    Send a transaction via Purple Flea Wallet API.
    Nonce is managed automatically — no NonceManager needed.
    """
    async with aiohttp.ClientSession() as session:
        async with session.post(
            "https://purpleflea.com/wallet-api/send",
            headers={"Authorization": f"Bearer {api_key}"},
            json={
                "chain":     chain,         # "ETH", "MATIC", "BNB", "AVAX"
                "to":        to,
                "value":     str(value_wei),
                "data":      data,
                "gas_mode":  "auto",        # API handles gas estimation
                "priority":  "normal",     # "slow" | "normal" | "fast" | "urgent"
            },
        ) as resp:
            resp.raise_for_status()
            result = await resp.json()
            # result = {"tx_hash": "0x...", "nonce": 47, "status": "submitted"}
            return result


# The API handles all nonce allocation, gas estimation, and retry logic.
# Your agent just calls send_via_purpleflea() for each transaction.