memserver.py

import asyncio
import os
import queue
import threading

from config import get_protocol, get_timestamp_seconds, get_config
from hashing import blake2b_hash
from ops.account_ops import get_account, get_finalized_height
from ops.block_ops import get_block_ends_info
from ops.data_ops import sort_list_dict, get_home
from ops.key_ops import load_keys
from ops.message_pool import MessagePool
from ops.transaction_ops import (
    validate_single_spending,
    validate_transaction,
    sort_transaction_pool,
    validate_txid

)
from ops import kv_ops   # tx-index oracle: an already-mined txid can never re-enter the mempool
from versioner import read_version


class MemServer:
    """storage thread for core.py, also accessed by most other threads, serves mostly as data storage"""

    def __init__(self, logger):
        """Assemble the node's ENTIRE shared runtime state in one place: keys, config (env vars
        override config.json for every headless knob), tx/message pools, pacing state, and the
        persisted safety floors (finalized_height). Constructed ONCE at startup and then shared by
        every loop thread. Cross-config invariants — notably max_rollbacks < finality_depth <
        EPOCH_LENGTH, which makes the epoch-beacon anchor un-reorgable — are asserted HERE so a
        mis-set config fails loudly at boot instead of silently disabling a protection later."""
        self.logger = logger
        self.logger.info("Starting MemServer")

        self.purge_peers_list = []

        self.start_time = get_timestamp_seconds()
        self.keydict = load_keys()
        self.config = get_config()
        # the handshake protocol number is CONSENSUS-ADJACENT and comes from the CODE, never from a
        # per-node config file (a stale persisted value silently kept old-rules peers admitted)
        self.protocol = get_protocol()
        self.private_key = self.keydict["private_key"]
        self.public_key = self.keydict["public_key"]
        self.address = self.keydict["address"]
        self.server_key = self.config["server_key"]
        # MEMPOOL LOCK (audit): transaction_pool is read-modify-REPLACED by the core loop while HTTP
        # executor threads and the peer loop append via merge_transaction — an append landing between
        # another thread's snapshot-read and list reassignment was silently LOST (a wallet got "Success"
        # for a tx that then never existed). Every mutation must hold this lock; reads of a single list
        # reference (e.g. hashing a .copy()) stay lock-free.
        # SINGLE MEMPOOL (2026-07): the old three-tier user_tx_buffer -> tx_buffer -> transaction_pool
        # cascade is collapsed to ONE pool. A submitted tx is validated and enters transaction_pool
        # directly (no staged promotion), and a tx already MINED (its txid is in the on-chain tx-index)
        # can never re-enter — merge_transaction, the producer filter, and verify_block all reject an
        # already-mined txid, so an IDENTICAL transaction (same content -> same txid) is impossible to
        # reintroduce or re-mine. A tx that misses its max_block simply expires; the wallet re-submits a
        # fresh tx (new nonce -> new txid) on the user's action, never silently re-injected.
        self.mempool_lock = threading.RLock()
        self.pool_gen = 0                 # bumped on EVERY pool mutation (see transaction_pool property)
        self._txid_set_cache = None        # (pool_gen, {txid}) — O(1) duplicate checks between mutations
        self.transaction_pool = []
        # GOSSIP REJECT CACHE {txid: retry_after_ts}: bodies fetched during set reconciliation that
        # merge_transaction REFUSED (expired target, cross-fork max_block, invalid). Without it a
        # divergent peer's pool hash never matches ours, so the SAME rejected bodies were re-fetched
        # and re-rejected EVERY ~1s peer pass, forever (observed against a wedged cross-fork peer:
        # its whole pool re-downloaded once a second, every tx failing "Target block too low").
        # LOCAL policy, bounded TTL — a tx that becomes valid later (e.g. funded account) is simply
        # retried after the cooldown; user submits via /submit_transaction never consult this.
        self._tx_reject_cache = {}
        self.message_pool = MessagePool()   # off-chain E2E message pool (doc/messaging.md); never block-bound
        # PERSIST the message pool across restarts — it is off-chain + ephemeral, so a plain node restart
        # (systemctl restart / redeploy) otherwise silently dropped every undelivered DM + published prekey.
        self.message_pool_path = f"{get_home()}/message_pool.dat"
        try:
            self.message_pool.load(self.message_pool_path, get_timestamp_seconds())
        except Exception:
            pass   # a corrupt/absent pool file is fine — a fresh empty pool is always valid
        self.peer_buffer = []
        # ONE of our addresses (the one we advertise), not our identity: this host is dual-stack and
        # peers list the other family back to us. Any "is this peer us?" check must use
        # ops.peer_ops.own_ips(), never `== memserver.ip` (ghost self-peer, 2026-09-06).
        self.ip = self.config["ip"]
        self.port = self.config["port"]
        self.terminate = False
        self.heavy_refresh_interval = 360

        # Target seconds between blocks (local production pacing — NOT consensus; verify only checks
        # timestamp <= now). Default 10s. All nodes on a network should agree on this so the chain keeps a
        # steady cadence; keep it identical across the mesh (ideally promote to a protocol constant later).
        self.block_time = self.config.get("block_time") or 10
        self.mode = "init"   # production pacing state (core_loop._mode): init | building | produce
        self.since_last_block = 0

        self.unreachable = {}
        self.peers = []

        # PUSH-GOSSIP outbound queue (ops/gossip.py): a newly-accepted tx is enqueued here and the
        # gossip worker (nado._gossip_worker) fans it out to peers immediately, so mempools converge
        # in one hop instead of waiting for the txid-diff pull reconcile. Bounded + best-effort: if it
        # backs up (a flood), we drop the overflow — the pull reconcile is the correctness backstop.
        self.gossip_queue = queue.Queue(maxsize=10000)

        self.transaction_pool_hash = None
        self.upcoming_block_hash = None   # hash of the NEXT block's tx set (mature subset) — the determinism signal
        self.reported_uptime = self.get_uptime()

        self.emergency_mode = False
        self.pool_warmed = False     # production gate: True after the first completed pool reconcile
                                     # with a peer (or solo/timeout fallback) — see core_loop.normal_mode

        self.version = read_version()
        block_ends_info = get_block_ends_info(logger=logger)
        self.latest_block = block_ends_info["latest_block"]
        self.earliest_block = block_ends_info["earliest_block"]
        # MEMPOOL CAPS (LOCAL policy, non-consensus — get_byte_size is a rough sys.getsizeof(repr) estimate).
        # These were ONE fused constant (150000) that meant "150k txs" in the accept gate but "150000 BYTES"
        # (~146 KB) in cull_buffer — and 146 KB is SMALLER than one block's blob budget (MAX_BLOB_BYTES_PER_BLOCK
        # = 256 KB), so cull evicted blob txs before a block could ever fill to 256 KB. Split into:
        #   transaction_pool_max_txs   — the count gate (how many txs the mempool may hold), and
        #   transaction_pool_max_bytes — the cull byte budget: MUST exceed a full block's blobs (so a block can
        #                                always fill) and stay under MAX_PEER_BODY (8 MiB, the /transaction_pool
        #                                fetch cap) so the pool stays transferable between peers. 4 MiB = 16
        #                                full blocks of blobs, well under the 8 MiB wire cap.
        # RAISED for INLINE SETTLE PROOFS. The stated invariant is "must exceed a full block's blobs, stay
        # under MAX_PEER_BODY" — both sides moved: a settle proof now rides INSIDE the tx (~120 MiB) rather
        # than through DA, which cannot deliver it on a fleet where only one node runs a DA store. A 4 MiB
        # cull budget would evict the one transaction the whole settlement path exists to carry. Keyed to
        # the same ceiling as the wire caps (protocol.MAX_INLINE_TX_BYTES) and still below MAX_PEER_BODY, so
        # /transaction_pool stays transferable.
        self.transaction_pool_max_txs = 150000
        from protocol import MAX_INLINE_TX_BYTES as _MAX_INLINE_TX
        self.transaction_pool_max_bytes = max(4 * 1024 * 1024, _MAX_INLINE_TX + (4 * 1024 * 1024))
        self.force_sync_ip = None
        self.rollbacks = 0
        self.can_mine = False

        _mp = self.config.get("min_peers")  # respect an explicit 0 (solo mode); `or 5` would force 5
        self.min_peers = 5 if _mp is None else _mp
        self.peer_limit = self.config.get("peer_limit") or 24
        self.max_rollbacks = self.config.get("max_rollbacks") or 10
        # ENFORCED FINALITY (#17, security step 1): a persisted monotonic finalized_height floor that
        # rollback_one_block REFUSES to cross (FinalityViolation). The ordering invariant below makes
        # the epoch-beacon anchor un-reorgable (a live epoch's anchor can never be reorged out) and
        # bounds 51%/long-range rollback. It fails loudly at startup so a mis-set config can't silently
        # disable the protection. (Stake-weighted fork-choice that bounds reorg COST is steps 2-3.)
        from protocol import EPOCH_LENGTH, FINALITY_DEPTH
        self.finality_depth = self.config.get("finality_depth") or FINALITY_DEPTH
        # CONFIG MIGRATION (2026-07-20): the 2026-07-19 finality widening (12 -> 45) shipped as a code
        # default, but a config file WRITTEN before it pins the old pair forever — /update never touches
        # private/, and no shell exists on some fleet boxes. A node left at depth 12 re-exposes the
        # 72s-partition freeze the widening prevents (and its duty reveals violate the new window on
        # every current peer), so an explicitly-NARROWER depth is lifted to the protocol constant and
        # persisted; max_rollbacks rides along per the widening rationale (a cap below the window cannot
        # traverse it). An operator pinning a DEEPER depth is left alone.
        if self.finality_depth < FINALITY_DEPTH:
            self.logger.warning(f"config migration: finality_depth {self.finality_depth} predates the "
                                f"{FINALITY_DEPTH} widening — lifting (and max_rollbacks -> 40)")
            self.finality_depth = FINALITY_DEPTH
            self.max_rollbacks = max(self.max_rollbacks, 40)
            try:
                from config import update_config
                update_config({"finality_depth": FINALITY_DEPTH, "max_rollbacks": self.max_rollbacks})
            except Exception as _e:
                self.logger.warning(f"config migration could not persist (applies in-memory anyway): {_e}")
        assert self.max_rollbacks < self.finality_depth < EPOCH_LENGTH, (
            f"need max_rollbacks ({self.max_rollbacks}) < finality_depth ({self.finality_depth}) "
            f"< EPOCH_LENGTH ({EPOCH_LENGTH}) for enforced-finality safety")
        # in-memory mirror of the persisted floor (advanced by core_loop.incorporate_block)
        self.finalized_height = get_finalized_height()
        # FFG (#6): the stake-attested finalized checkpoint height (observability; <= finalized_height,
        # which is the deeper time-based floor that bounds rollback). Updated by core_loop.maybe_attest.
        self.ffg_finalized = 0
        # RANDAO (#7): this validator's locally-held secrets {target_epoch: secret}, committed in E-2
        # and revealed in E-1. PERSISTED to private/randao_secrets.json (node-local, gitignored, beside
        # the keys): they were in-memory only ("a wasted commit — harmless") until the integrated
        # auto-updater made restarts ROUTINE — an update wave landing between a commit (E-2) and its
        # reveal (E-1) wasted the commit every time; on 2026-07-20 an evening of deploys produced four
        # straight epochs of reveal=False from this node. Saved on every new commit + pruned to live
        # epochs in core_loop.maybe_epoch_duty.
        self.randao_secrets = {}
        try:
            import json as _json, os as _os
            _rs_path = f"{get_home()}/private/randao_secrets.json"
            if _os.path.isfile(_rs_path):
                with open(_rs_path) as _rs:
                    self.randao_secrets = {int(k): v for k, v in _json.load(_rs).items()
                                           if isinstance(v, str) and v}
                if self.randao_secrets:
                    self.logger.info(f"Recovered {len(self.randao_secrets)} unrevealed RANDAO secret(s) "
                                     f"for epoch(s) {sorted(self.randao_secrets)}")
        except Exception as _e:
            self.logger.warning(f"could not load persisted RANDAO secrets (continuing fresh): {_e}")
        # Fast bootstrap is snapshot sync (ops/snapshot_ops.py) — quorum/checkpoint-gated, never a
        # validation-skipping bypass. Do NOT add one (it enables forged-tx injection).
        # AUTO-BOND (non-consensus): route this % of newly-mined spendable earnings straight into
        # bonded stake, unattended (core_loop.maybe_auto_bond). Defaults to AUTO_BOND_DEFAULT_PERCENT
        # (80) when unset — a fresh node joins the bonded lane hands-free; 0 = off. Source order:
        # NADO_AUTO_BOND_PERCENT env (handy for headless/systemd) overrides config["auto_bond_percent"].
        import os as _os
        from protocol import AUTO_BOND_DEFAULT_PERCENT as _AB_DEFAULT
        _ab = _os.environ.get("NADO_AUTO_BOND_PERCENT")
        if _ab is None:
            _ab = self.config.get("auto_bond_percent", _AB_DEFAULT)
        try:
            self.auto_bond_percent = max(0, min(100, int(_ab)))
        except (TypeError, ValueError):
            self.auto_bond_percent = _AB_DEFAULT

        # AUTO-COLLECT the presence dividend, unattended (core_loop.maybe_auto_collect) — DEFAULT ON. Only an
        # OPEN-lane member accrues one, so a bonded-only node is a no-op. NADO_AUTO_COLLECT env overrides config.
        def _flag(env, cfg, default):
            """Boolean knob resolved env var > config[cfg] > default; '0'/'false'/'no'/'off'
            (any case) mean False, anything else True — so systemd Environment= lines and
            hand-edited config values both behave predictably."""
            v = _os.environ.get(env)
            v = self.config.get(cfg, default) if v is None else v
            return str(v).strip().lower() not in ("0", "false", "no", "off")
        self.auto_collect_dividend = _flag("NADO_AUTO_COLLECT", "auto_collect_dividend", True)
        # rolling-mode tx-history window (0 = the protocol floor); read here so the core loop's prune pass
        # sees an operator override without re-reading config on every block.
        # `os` was never imported here, so this raised NameError on EVERY boot — swallowed by the bare
        # `except Exception` below, which quietly pinned the retention to 0. Neither the env var nor the
        # config key had any effect on any node, and nothing said so. (Found by test_no_undefined_names,
        # which is exactly the omission that test exists for.)
        try:
            self.tx_history_retention_blocks = int(
                os.environ.get("NADO_TX_HISTORY_RETENTION")
                or self.config.get("tx_history_retention_blocks", 0) or 0)
        except Exception:
            self.tx_history_retention_blocks = 0
        # Latest conservation-invariant reconciliation (ops/invariants.py), refreshed by the core loop's
        # periodic duty and served read-only at /invariants. None until the first check runs. Purely a
        # detector cache — nothing consensus reads it.
        self.invariant_report = None
        # Latest self-update capability diagnosis (ops/self_update.ensure_updatable), refreshed at boot and
        # daily, advertised in /status as update_capable so a lagging/unfixable node is visible fleet-wide.
        self.updatability = None
        # Optional exec-layer view for the escrow invariants. A node with no exec side leaves this None and
        # only the L1 supply invariant runs.
        self.exec_state_view = None
        # AUTO-REGISTER + renew the open-lane PoSW lease, unattended — DEFAULT ON.
        #
        # It was opt-in ("a headless node should not silently join, and Sybil-load, the open lane"), and the
        # cost of that default was measured on betanet-2: of 8 fleet nodes, TWO were running, validating and
        # relaying for the whole life of the chain while earning EXACTLY ZERO — 0.00 produced, fidelity 0,
        # registered 0 — purely because nobody had set an environment variable on them. Running a node and
        # getting nothing is not a safe default, it is a silent misconfiguration that looks like working.
        #
        # The Sybil worry the old default guarded against is now handled where it belongs, in consensus
        # rather than in a flag: POSW_ENTRY_MULT makes creating an identity cost 32x a renewal, and
        # FIDELITY_MIN_GAP_EPOCHS stops the continuity ramp being farmed. Neither existed when this default
        # was chosen. Note also that auto-register was never the Sybil lever anyway — it registers ONE
        # identity, this node's own key; a Sybil does not need it.
        #
        # NADO_AUTO_REGISTER=0 (or config auto_register:false) opts a node out.
        self.auto_register = _flag("NADO_AUTO_REGISTER", "auto_register", True)

        # AUTO-VOTE on treasury proposals paying a WHITELISTED recipient — DEFAULT ON, whitelist-restricted.
        #
        # WHY THIS EXISTS AT ALL: the browser wallet has auto-voted since the feature shipped, but treasury
        # quorum is counted in BONDED SHARES, and browser miners hold almost none. Measured on betanet-2:
        # 108 of 117 open miners have ZERO voting shares, so their auto-vote is a no-op, while all 42 shares
        # sit with 9 bonded node operators whose software never voted at all. Quorum is 28 of 42 — literally
        # unreachable, which is why the treasury held 109 NADO with zero proposals ever paid. Whitelisting
        # was decorative until the side holding the weight also votes.
        #
        # The whitelist is the safety property, not the flag: a node will only ever auto-approve a spend to
        # a listed recipient. The default is the reserved `faucet` escrow — keyless, public-good, and
        # impossible to redirect — so the shipped behaviour cannot move funds to anyone's address. Operators
        # widen it deliberately (NADO_AUTO_VOTE_ALLOW="faucet,<addr>") or switch it off entirely.
        self.auto_vote = _flag("NADO_AUTO_VOTE", "auto_vote", True)
        _allow = _os.environ.get("NADO_AUTO_VOTE_ALLOW")
        if _allow is None:
            _allow = self.config.get("auto_vote_allow", "faucet")
        self.auto_vote_allow = [a.strip().lower() for a in str(_allow).split(",") if a.strip()]

        # ROLLING MODE (non-consensus): archive=True (default) keeps ALL block bodies; False runs a
        # pruned/rolling node that drops bodies older than history_retention_blocks (state + indexes
        # kept). NADO_ARCHIVE=0/false selects rolling mode headless. See doc/rolling-mode-and-da.md.
        from protocol import HISTORY_RETENTION_BLOCKS as _HRB
        _arch = _os.environ.get("NADO_ARCHIVE")
        if _arch is not None:
            self.archive = _arch.strip().lower() not in ("0", "false", "no", "off")
        else:
            # ROLLING BY DEFAULT — must match config.py's "archive" default, and it is this fallback (not
            # the config file) that decides for every node whose config predates the key. Archive costs a
            # measured ~47.6 GB/year of block bodies; a node that fills its disk stops UPDATING, not just
            # archiving. Existing configs with an explicit "archive": true keep archiving, untouched.
            self.archive = bool(self.config.get("archive", False))
        try:
            _hrb = int(_os.environ.get("NADO_HISTORY_RETENTION_BLOCKS")
                       or self.config.get("history_retention_blocks", 0) or 0)
        except (TypeError, ValueError):
            _hrb = 0
        self.history_retention_blocks = _hrb if _hrb > 0 else _HRB

        # per-IP registration budget + identity cap RETIRED at gen 25 (device attestation); knobs gone.
    def ban_peer(self, peer):
        """Queue a misbehaving/unreachable peer for purge (deduplicated against both the purge list
        and the already-unreachable set). Seed peers are EXEMPT — never exiled, always retried —
        because they are the weak-subjectivity anchor (see below)."""
        # Operator seeds are the weak-subjectivity anchor — NEVER exile them. A transient blip (e.g. the
        # seed restarting) would otherwise drop it into the 1-hour unreachable ban, its heavy tip vanishes
        # from the pool, and the node falls back to whatever stalled/forked peer is left. A seed is always
        # retried instead.
        from ops.peer_ops import seed_peers
        if peer in seed_peers():
            return
        if peer not in self.purge_peers_list and peer not in self.unreachable:
            self.purge_peers_list.append(peer)

    # HASH CACHES for the two per-second consensus signals below. Key = pool_gen, the counter the
    # transaction_pool property setter bumps on EVERY reassignment (merge_transaction's post-append
    # sort, purge, drain, cull) — see core_loop's "ASSIGN ONLY ON CHANGE" note: an earlier key on the
    # list object + its length never hit under load, because the core loop reassigned the pool every
    # pass. An unchanged pool_gen proves the tx set is unchanged and the cached hash is exact. This
    # turns the old O(pool)·sort + canonical-serialize EVERY SECOND into O(1) between pool changes —
    # the whole-mempool rehash was the classic hot-loop serialization cost at mempool scale.
    _pool_hash_cache = None            # (pool_gen, hash)
    _upcoming_hash_cache = None        # (pool_gen, parent_hash, kv_write_gen, hash)

    # The pool is a property so EVERY reassignment (merge append path, purges, the core loop's
    # drain/cull swaps) bumps pool_gen — the one content-change signal all pool-derived caches key on.
    # (Object identity is NOT a safe key: CPython reuses freed list addresses, and in-place appends
    # keep identity while changing content. In-place mutation sites bump pool_gen explicitly.)
    @property
    def transaction_pool(self):
        return self._transaction_pool

    @transaction_pool.setter
    def transaction_pool(self, value):
        self._transaction_pool = value
        self.pool_gen += 1

    _inflight_txids = set()
    _inflight_lock = threading.Lock()

    def _pool_txid_set(self):
        """set of pooled txids at the current pool_gen — rebuilt only after a mutation, so the
        per-second re-gossip storm (peers re-serve their whole pool) dedups in O(1) per tx instead
        of an O(pool) deep-compare over the full posw payloads."""
        c = self._txid_set_cache
        if c is None or c[0] != self.pool_gen:
            c = (self.pool_gen, {t.get("txid") for t in self._transaction_pool})
            self._txid_set_cache = c
        return c[1]

    def get_transaction_pool_hash(self) -> [str, None]:
        """blake2b of the SORTED transaction pool (None when empty). Sorting first makes the hash
        canonical — two nodes holding the same tx set report the same hash regardless of arrival
        order — which is what lets the consensus loop majority-vote on pool hashes instead of
        shipping full pools around. Hashes a copy so a concurrent merge can't mutate mid-sort.
        Cached per pool object+length (see cache note above) — a pure function of the tx set."""
        pool = self.transaction_pool
        if not pool:
            return None
        cached = self._pool_hash_cache
        gen = self.pool_gen
        if cached is not None and cached[0] == gen:
            return cached[1]
        pool_hash = blake2b_hash(sort_transaction_pool(pool.copy()))
        self._pool_hash_cache = (gen, pool_hash)
        return pool_hash

    def get_upcoming_block_hash(self):
        """blake2b of the NEXT block's content ON TOP OF OUR TIP: parent hash + next height + the mature,
        target-height tx subset (match_transactions_target). This is EXACTLY what block determinism / the
        fast-forward depend on — two nodes at the same tip agree here iff they will build the identical
        next block, INCLUDING an empty one (a produced block always has a hash). Unlike the whole-pool
        hash it excludes immature (min_block not reached) and future-targeted txs that won't be in the
        next block. parent+height make it tip-specific, so nodes on a different tip correctly don't match.
        NEVER None (an empty next block still hashes) — so a peer's absence of eligible txs is a real,
        comparable signal, not a null that poisons the majority.
        Cached per (pool object+length, tip hash, committed-write generation): the match also reads
        the on-chain mined-txid index, and the write generation invalidates on ANY commit, so the
        cache can never outlive the state it was derived from."""
        from ops.block_ops import match_transactions_target
        parent = self.latest_block
        pool = self.transaction_pool
        cached = self._upcoming_hash_cache
        key = (self.pool_gen, parent["block_hash"], kv_ops.write_generation())
        if cached is not None and cached[0:3] == key:
            return cached[3]
        next_height = parent["block_number"] + 1
        matched = match_transactions_target(transaction_list=pool.copy(),
                                            block_number=next_height, logger=self.logger) if pool else []
        if matched is False:                              # a match error -> treat as empty
            matched = []
        upcoming = blake2b_hash([parent["block_hash"], next_height, sort_transaction_pool(matched)])
        self._upcoming_hash_cache = (*key, upcoming, [t.get("txid") for t in matched])
        return upcoming

    def get_next_block_txids(self):
        """(tip_hash, next_height, [txid, ...]) — the exact tx set get_upcoming_block_hash hashes, served
        by /next_block_txids for the PRE-ASSEMBLY RECONCILE (reconcile_next_block_set). Same cache."""
        self.get_upcoming_block_hash()
        c = self._upcoming_hash_cache
        return c[1], self.latest_block["block_number"] + 1, list(c[4])

    def reconcile_next_block_set(self, peers, timeout=1.5) -> dict:
        """PRE-ASSEMBLY TX-SET RECONCILE (2026-09-01, leaderless assembly touch-up; doc/leaderless-assembly.md).

        Blocks are a pure function of the mempool and EVERY node assembles every block, so a same-height
        split needs exactly one thing: a tx that reached one assembler and not another before the slot.
        The pull reconcile (merge_remote_transactions) converges pools in ~1 s passes, and the status-pool
        agreement signal is polled every ~10 s — both slower than a 6 s block. This is the direct check,
        run ONCE per tip right before we build: ask peers what THEY would put in the next block on THIS tip
        (/next_block_txids, ~64 B per tx, 1.5 s budget), fetch only the bodies we lack from a peer that has
        them (/transactions_by_id), merge them through the ordinary admission path, and only then assemble.
        Symmetric: every peer does the same against us, so the whole mesh converges on the UNION of its
        next-block sets within the slot instead of racing on the differences. Bounded and best-effort: a
        peer that does not answer in time simply sits out this pass; nothing here is trusted (every
        fetched tx is fully validated by merge_transaction) and nothing here can stall production beyond
        the budget. Returns counters for the log/telemetry."""
        from compounder import compound_get_next_block_txids, post_txs_by_id
        peers = list(peers)[:16]
        if not peers:
            return {"peers": 0}
        tip = self.latest_block["block_hash"]
        ours = set(self.get_next_block_txids()[2])
        answers = asyncio.run(compound_get_next_block_txids(peers, self.port, self.logger,
                                                            asyncio.Semaphore(16), timeout))
        same = {p: d for p, d in answers.items() if d.get("tip") == tip}
        local = self._pool_txid_set()
        want_by_peer, claimed = {}, set()
        for p, d in same.items():
            want = [i for i in d["txids"]
                    if isinstance(i, str) and len(i) <= 64 and i not in local and i not in claimed
                    and kv_ops.tx_get(i) is None]
            if want:
                want_by_peer[p] = want[:200]
                claimed.update(want[:200])
        fetched = 0
        if want_by_peer:
            async def _fetch():
                sem = asyncio.Semaphore(8)
                coros = [post_txs_by_id(p, self.port, ids, self.logger, [], sem) for p, ids in want_by_peer.items()]
                try:
                    res = await asyncio.wait_for(asyncio.gather(*coros, return_exceptions=True), timeout=3.0)
                except asyncio.TimeoutError:
                    return []
                return [tx for r in res if isinstance(r, list) for tx in r]
            for tx in asyncio.run(_fetch()):
                if not isinstance(tx, dict) or self._is_proof_settle(tx):
                    continue
                r = self.merge_transaction(tx)
                if isinstance(r, dict) and r.get("result") and r.get("message") == "Success":
                    fetched += 1
        agreed = sum(1 for d in same.values() if set(d["txids"]) == ours)
        return {"peers": len(answers), "same_tip": len(same), "agreed": agreed,
                "missing": sum(len(v) for v in want_by_peer.values()), "fetched": fetched}

    def get_uptime(self) -> int:
        """Whole seconds this node process has been up (NOT system uptime) — refreshed into
        reported_uptime by the core loop and shared with peers via /status."""
        return get_timestamp_seconds() - self.start_time

    # per-peer/per-pass bound on bodies fetched during set reconciliation; the server side caps a
    # /transactions_by_id request at the same figure. The remainder arrives on the next 1s pass.
    _RECONCILE_MAX_IDS = 1000

    @staticmethod
    def reject_cooldown_s(message, funder_in_pool):
        """How long a REFUSED gossip tx cools before re-fetch — graded by what the refusal MEANS.

        A flat 60 s cooled every refusal equally, and that was a fork driver: "Empty account" is a spend
        from a fresh address whose FUNDING tx is still in the mempool — the account exists only once the
        funding MINES (~TX_INCLUSION_DELAY blocks), so nodes that saw the spend early refused + cooled it
        for 60 s while nodes that saw it late admitted it. Divergent pools for up to ~10 blocks, and
        deterministic production turns a divergent pool straight into a split (the 62655/62895 class).

        TERMINAL refusals (the tx can never become valid — bad bytes, bad signature, already mined,
        superseded, out-of-window target) keep the long cooldown: re-fetching them is pure waste.
        TRANSIENT refusals (account not funded YET, mempool full) cool briefly — 2 block times — so the
        pool re-converges within the same window the funding needs to mine. And an "Empty account" whose
        funder we can SEE in our own pool does not cool at all: the very next reconcile pass after the
        funding mines admits it, which is the earliest any node can."""
        TERMINAL = ("Malformed transaction", "Invalid txid", "Invalid signature", "Already mined",
                    "Superseded", "Target block too high", "Target block too low")
        if any(t in str(message) for t in TERMINAL):
            return 60
        if "Empty account" in str(message) and funder_in_pool:
            return 0
        # EXEC-SETTLE SKEW: dividend claims (and their collect blobs) validate at admission against the
        # receiving node's OWN settled exec root, which lags differently per node — a claim proven
        # against a root this node has not settled YET was refused + cooled 12s repeatedly, delaying
        # delivery for minutes (measured: fork seed h68851/h68921 — the majority never held a 3-minute-old
        # claim). The refusal resolves the moment OUR exec settles; cool 0 so the first reconcile after
        # that admits it, the earliest any node can.
        if "not proven against the settled" in str(message) or "no settled execution-layer root" in str(message):
            return 0
        return 12

    @staticmethod
    def _is_proof_settle(tx) -> bool:
        d = tx.get("data") if isinstance(tx, dict) else None
        return tx.get("recipient") == "settle" and isinstance(d, dict) and ("proof" in d or "proof_da" in d)

    def _queue_proof_merge(self, tx, user_origin):
        """Hand a proof-bearing settle to the single proof worker (dedup by txid: one verification per tx)."""
        import queue, threading
        if not hasattr(self, "_proof_q"):
            self._proof_q, self._proof_inflight = queue.Queue(), set()
            def _worker():
                while True:
                    tx, uo = self._proof_q.get()
                    try:
                        self.merge_transaction(tx, uo)
                    except Exception as e:
                        self.logger.error(f"proof worker: {e}")
                    finally:
                        self._proof_inflight.discard(tx.get("txid"))
            threading.Thread(target=_worker, name="proof_verify", daemon=True).start()
        txid = tx.get("txid")
        if txid in self._proof_inflight or txid in self._pool_txid_set():
            return
        self._proof_inflight.add(txid)
        self._proof_q.put((tx, user_origin))

    # ---- MEMPOOL PERSISTENCE ----------------------------------------------------------------------------
    # A restart used to drop every pooled transaction on the floor: a client had been told "accepted" and
    # the tx simply never existed again (2026-08-29: an order posted seconds before a self-update restart
    # vanished). The pool is written to disk on shutdown and every few seconds, and re-merged on boot
    # through the SAME validation as any gossiped tx — nothing is trusted from disk, only remembered.
    @property
    def pool_path(self):
        from ops.data_ops import get_home
        return f"{get_home()}/mempool.json"

    # ---- EXPIRY ---------------------------------------------------------------------------------------
    # A tx whose landing window has closed (max_block at or below the tip, or beyond TX_LANDING_WINDOW)
    # can never be mined. Until 2026-09-04 it was dropped only on the core loop's PRODUCE path — and a
    # node that is behind sits inside emergency_mode's internal while-loop and never produces, so the
    # relay held a 10 MiB dead settle tx for 1,500 blocks, served it to four peers ~1/s through
    # /transactions_by_id (they cannot admit it, so reconcile never converged), re-dumped it to disk
    # every 5 s, and spent 67 % of the GIL serialising it — too slow to ever catch up and prune. The pool
    # OWNER prunes, on the peer loop's 1 s clock (every mode), and the peer-facing reads never see a
    # dead tx even between prunes.
    def _tx_can_land(self, tx) -> bool:
        from protocol import TX_LANDING_WINDOW
        try:
            h = self.latest_block["block_number"]
            return h < tx["max_block"] < h + TX_LANDING_WINDOW
        except Exception:
            return True                      # no tip yet / odd shape: never silently hide, admission decides

    def live_pool(self):
        """The pool minus anything that can no longer land — what peers are served and what is persisted."""
        return [t for t in self.transaction_pool if self._tx_can_land(t)]

    def prune_expired_pool(self) -> int:
        """Drop txs whose landing window has closed. Snapshot-filter-reassign under the mempool lock (vs
        concurrent merge_transaction appends); the pool is reassigned only when something actually went.
        Returns the number dropped."""
        try:
            height = self.latest_block["block_number"]
        except Exception:
            return 0
        with self.mempool_lock:
            before = self.transaction_pool
            kept = [t for t in before if self._tx_can_land(t)]
            dropped = len(before) - len(kept)
            if dropped:
                self.transaction_pool = kept
        if dropped:
            self.logger.warning(f"Pruned {dropped} expired tx(s) from the pool at height {height}")
        return dropped

    def save_pool(self, force=False):
        import time as _t
        now = _t.time()
        if not force and now - getattr(self, "_pool_saved_at", 0) < 5:
            return 0
        self._pool_saved_at = now
        import json as _json
        import threading as _th
        try:
            txs = self.live_pool()                 # never persist a tx that can no longer land
            # UNCHANGED POOL ⇒ NO WRITE, AND NEVER ON THE CALLER'S THREAD. Six pending 10 MiB settle proofs made
            # this a 60 MiB json.dump every 5 s on the PEER loop (py-spy 2026-09-07 13:29: 7/10 samples here),
            # which is the loop that syncs blocks — the relay crawled 100+ blocks behind the fleet. The set of
            # txids is the identity of the pool; serialise only when it moved, and on a daemon thread.
            sig = tuple(sorted(str(t.get("txid", "")) for t in txs if isinstance(t, dict)))
            if sig == getattr(self, "_pool_sig", None):
                return len(txs)
            w = getattr(self, "_pool_writer", None)
            if w is not None and w.is_alive():
                if not force:
                    return len(txs)                # a write is in flight; the next tick re-checks the signature
                w.join(timeout=30)                 # shutdown: let it finish, then write the final state synchronously
            self._pool_sig = sig

            def _write(txs=txs, sig=sig):
                try:
                    tmp = self.pool_path + ".tmp"
                    with open(tmp, "w") as f:
                        _json.dump(txs, f)
                    os.replace(tmp, self.pool_path)
                except Exception as e:
                    self._pool_sig = None          # retry next tick
                    self.logger.error(f"mempool persist failed: {e}")
            if force:
                _write()
            else:
                self._pool_writer = _th.Thread(target=_write, daemon=True, name="mempool-persist")
                self._pool_writer.start()
            return len(txs)
        except Exception as e:
            self.logger.error(f"mempool persist failed: {e}")
            return 0

    def load_pool(self):
        """Re-merge the last persisted pool. Every tx goes through merge_transaction (full validation,
        min/max_block window, spending) so a stale or forged file can only ever be refused."""
        import json as _json
        try:
            with open(self.pool_path) as f:
                txs = _json.load(f)
        except FileNotFoundError:
            return 0
        except Exception as e:
            self.logger.error(f"mempool restore: unreadable file ignored: {e}")
            return 0
        kept = 0
        for tx in txs if isinstance(txs, list) else []:
            try:
                if self._is_proof_settle(tx):
                    self._queue_proof_merge(tx, False); continue
                r = self.merge_transaction(tx, False)
                kept += 1 if isinstance(r, dict) and r.get("result") else 0
            except Exception:
                pass
        self.logger.info(f"mempool restored: {kept} of {len(txs) if isinstance(txs, list) else 0} persisted transactions re-admitted")
        return kept

    def merge_remote_transactions(self, user_origin=False, skip_pool_peers=()) -> None:
        """MEMPOOL SET RECONCILIATION (replaces the full-pool download): for each peer whose
        advertised pool hash differs from ours (skip_pool_peers filters the identical ones), fetch
        its txid LIST (/transaction_ids, ~64B/tx), diff against what we hold + what is already
        MINED, and fetch ONLY the missing bodies (/transactions_by_id). The old path re-downloaded
        every divergent peer's ENTIRE pool every second — O(peers × pool) bandwidth for mostly-known
        data; this is O(peers × ids) + O(genuinely missing bodies), a ~100x cut with ~7KB ML-DSA
        txs. Each missing txid is claimed from ONE peer per pass (no duplicate downloads)."""
        pool_peers = [p for p in self.peers if p not in skip_pool_peers]
        if pool_peers:
            missing = asyncio.run(self._fetch_missing_remote_txs(pool_peers))
            now = get_timestamp_seconds()
            for tx in missing:
                # A PROOF-BEARING SETTLE is verified on its own worker, one at a time, never inline: its
                # STARK verification runs for minutes in Python, and inline it froze this thread's status
                # pass (2026-08-29, 70 blocks behind while believing it led the mesh). The worker calls the
                # very same merge_transaction, so admission rules are unchanged — only the thread differs.
                if self._is_proof_settle(tx):
                    self._queue_proof_merge(tx, user_origin)
                    continue
                result = self.merge_transaction(tx, user_origin)
                # REFUSED gossip body -> cool it down (see _tx_reject_cache): don't re-fetch the same
                # rejected tx from the same divergent peer every second. 60s TTL: a transient reason
                # (mempool full, account funded later) is retried after the cooldown.
                if isinstance(result, dict) and not result.get("result") and isinstance(tx.get("txid"), str):
                    _funder_seen = any(isinstance(t, dict) and t.get("recipient") == tx.get("sender")
                                       for t in self.transaction_pool)
                    _cool = self.reject_cooldown_s(result.get("message"), _funder_seen)
                    if _cool:
                        self._tx_reject_cache[tx["txid"]] = now + _cool
                    # REJECT TELEMETRY (/status recent_tx_rejects): fork seed h68261 was a 4-minute-old
                    # collect blob our pool simply did not hold — and nothing recorded whether it was
                    # refused (and why) or never delivered. A refused SYSTEM tx is a future fork; make
                    # every gossip refusal readable from outside. Ring of the last 20.
                    try:
                        self.recent_tx_rejects = ([{"txid": tx["txid"][:16],
                                                    "recipient": str(tx.get("recipient"))[:24],
                                                    "why": str(result.get("message"))[:60], "at": now}]
                                                  + getattr(self, "recent_tx_rejects", []))[:20]
                    except Exception:
                        pass
            # bounded: drop expired entries; hard-cap so a flood of unique invalid txids can't grow it
            if len(self._tx_reject_cache) > 20000:
                self._tx_reject_cache = {i: t for i, t in self._tx_reject_cache.items() if t > now}

    async def _fetch_missing_remote_txs(self, pool_peers) -> list:
        """ids from all divergent peers in parallel -> per-peer want-lists (deduped across peers,
        mined txids excluded) -> parallel bounded body fetches. A peer that cannot serve the
        reconciliation wire goes to the purge queue like any other failing peer (no legacy wire)."""
        from compounder import compound_get_tx_ids, post_txs_by_id
        semaphore = asyncio.Semaphore(50)
        ids_by_peer = await compound_get_tx_ids(pool_peers, self.port, self.logger,
                                                self.purge_peers_list, semaphore)
        if not ids_by_peer:
            return []
        local = {t.get("txid") for t in self.transaction_pool}
        claimed = set()
        plans = []
        _now = get_timestamp_seconds()
        for peer, ids in ids_by_peer.items():
            want = []
            for i in ids:
                if (isinstance(i, str) and len(i) <= 64 and i not in local and i not in claimed
                        and self._tx_reject_cache.get(i, 0) <= _now   # recently refused -> cooling down
                        and kv_ops.tx_get(i) is None):        # already MINED -> never re-fetch (the old flood)
                    want.append(i)
                    if len(want) >= self._RECONCILE_MAX_IDS:
                        break
            if want:
                claimed.update(want)
                plans.append((peer, want))
        if not plans:
            return []
        batches = await asyncio.gather(*[
            post_txs_by_id(peer, self.port, want, self.logger, self.purge_peers_list, semaphore)
            for peer, want in plans])
        out = []
        for batch in batches:
            if isinstance(batch, list):
                out.extend(batch)
        return out


    def maybe_watchtower_slash(self, transaction):
        """THE WATCHTOWER: automatic slashing on an OBSERVED FFG double-vote. Slash validation has been
        complete for a while (resolve_slash: both proof kinds, dedup, penalty) — but nothing ever
        SUBMITTED one, so punishment existed only if a human noticed and hand-built the proof tx.
        Accountability that depends on someone watching is not accountability.

        Every gossiped attest-bearing tx passes through here BEFORE admission (the equivocating second
        vote is typically REFUSED as a duplicate section, so hooking after admission would drop the
        evidence). We remember the last signed attest tx per (sender, epoch); a second one with a
        DIFFERENT target_hash is an irrefutable double-vote — build the proof from the two full signed
        txs and merge the fee-exempt slash tx into our own pool (gossip carries it from there). The
        memory is one tx per validator per epoch, pruned two epochs back — bounded by the committee size.

        Never raises into admission, and never fires on our own txs re-entering (same target_hash)."""
        try:
            if not isinstance(transaction, dict) or transaction.get("recipient") not in ("duty", "attest"):
                return
            d = transaction.get("data") or {}
            att = d.get("attest") if transaction.get("recipient") == "duty" else d
            if not isinstance(att, dict):
                return
            epoch, thash = att.get("target_epoch"), att.get("target_hash")
            sender = transaction.get("sender")
            if not isinstance(epoch, int) or isinstance(epoch, bool) or not thash or not sender:
                return
            if not hasattr(self, "_attest_watch"):
                self._attest_watch = {}
            prev = self._attest_watch.get((sender, epoch))
            if prev is None:
                self._attest_watch[(sender, epoch)] = transaction
                # prune: nothing older than 2 epochs back can still be slashed ahead of its dedup anyway
                for k in [k for k in self._attest_watch if k[1] < epoch - 2]:
                    self._attest_watch.pop(k, None)
                return
            prev_att = (prev.get("data") or {}).get("attest") if prev.get("recipient") == "duty" \
                else (prev.get("data") or {})
            if not isinstance(prev_att, dict) or prev_att.get("target_hash") == thash:
                return                                    # same vote re-gossiped: honest
            from protocol import TX_LANDING_WINDOW
            from ops.transaction_ops import verify_attestation_equivocation_proof, construct_slash_tx
            proof = {"attest_a": prev, "attest_b": transaction}
            if not verify_attestation_equivocation_proof(proof):
                return                                    # would not survive validation — do not spam
            self.logger.warning(f"WATCHTOWER: FFG double-vote by {sender[:16]}… at epoch {epoch} — "
                                f"submitting the slash")
            slash = construct_slash_tx(self.keydict, proof,
                                       self.latest_block["block_number"] + TX_LANDING_WINDOW - 5)
            self.merge_transaction(slash, user_origin=True)
        except Exception as e:
            self.logger.warning(f"watchtower error (non-fatal): {e}")

    def merge_transaction(self, transaction, user_origin=False) -> dict:
        """warning, can get stuck if not efficient"""
        from protocol import TX_LANDING_WINDOW
        # AUDIT FIX: a malicious peer can serve a /transaction_pool list with a malformed entry; the
        # pre-validation field accesses below (sender, max_block) would KeyError/TypeError and abort
        # the whole merge batch. Reject malformed txs up front so the rest of the batch still merges.
        if (not isinstance(transaction, dict) or "sender" not in transaction
                or not isinstance(transaction.get("max_block"), int)
                or isinstance(transaction.get("max_block"), bool)):
            return {"result": False, "message": "Malformed transaction"}

        # WATCHTOWER (before any refusal path — a refused duplicate attest section IS the evidence)
        self.maybe_watchtower_slash(transaction)

        # RE-GOSSIP FAST PATH: peers re-serve their whole pool every second, so the overwhelmingly
        # common case is a tx we already hold. O(1) txid-set check BEFORE any validation or DB read —
        # the old path deep-compared against the full 25 MiB pool (line: `transaction in pool`) and
        # re-hashed the 36 KiB posw body per duplicate, which alone saturated the GIL at fleet scale.
        # Success, not error: it IS present (same contract as the late "Already present" branch).
        _txid = transaction.get("txid")
        if isinstance(_txid, str) and _txid in self._pool_txid_set():
            return {"message": "Already present", "result": True}
        # IN-FLIGHT DEDUPE (2026-09-02): the pool-membership check above is blind to a tx whose FIRST
        # validation is still running — nine peers pushing the same settle tx within a second put nine
        # copies through full proof verification at once (see transaction_ops._SETTLE_VERIFY_GATE for what
        # that cost). The second arrival of an in-flight txid is acknowledged, not re-validated.
        if isinstance(_txid, str):
            with self._inflight_lock:
                if _txid in self._inflight_txids:
                    return {"message": "Already being validated", "result": True}
                self._inflight_txids.add(_txid)
        try:
            return self._merge_transaction_validated(transaction, user_origin, _txid)
        finally:
            if isinstance(_txid, str):
                with self._inflight_lock:
                    self._inflight_txids.discard(_txid)

    def _merge_transaction_validated(self, transaction, user_origin, _txid):
        """The validating half of merge_transaction (runs once per txid at a time — see the in-flight set)."""
        from protocol import TX_LANDING_WINDOW

        # Anti-DoS: hard-cap the mempool so a flood (incl. fee-exempt register/heartbeat spam) cannot
        # grow it unbounded and OOM the node. Pairs with the per-IP HTTP rate limiter. The lane cap
        # already stops spam from buying extra block share; this stops it taking the node down.
        # PERF: O(1) length check on the single pool — a flood already at the cap is rejected in O(1).
        if len(self.transaction_pool) >= self.transaction_pool_max_txs:
            return {"result": False, "message": "Mempool full"}

        # CHEAP BOUNDS FIRST (audit): the two integer max_block compares run before the LMDB
        # get_account read — a stale re-gossiped tx (the most common reject under load, since peers
        # re-serve their whole pool every second) must not cost a DB hit to reject.
        # `<=` (was `<`): the drain promotes only max_block STRICTLY greater than the tip
        # (merge_buffer block_min < target), so a tx targeting the current tip was acknowledged
        # "Success", could never be mined, and silently aged out — reject it up front instead.
        if transaction["max_block"] <= self.latest_block["block_number"]:
            msg = {"result": False,
                   "message": f"Target block too low"}
            return msg

        elif transaction["max_block"] > self.latest_block["block_number"] + TX_LANDING_WINDOW:
            msg = {"result": False,
                   "message": f"Target block too high"}
            return msg

        # USER-ORIGIN PROPAGATION GUARD (door-level, so it can never diverge pools: a tx refused at its
        # OWN entry door never exists anywhere). Browser wallets running a stale cached interface.js
        # still build dividend claims with no min_block — one seeded the h67961 4v5 split, another
        # forked h68851 AFTER the JS fix shipped (the fix cannot reach a client that has not refreshed).
        # Gossip-origin txs are exempt: an already-admitted legacy tx must keep relaying identically on
        # every node.
        elif (user_origin and transaction.get("recipient") == "dividend_withdraw"
                and int(transaction.get("min_block", 0) or 0)
                        < self.latest_block["block_number"] + 4):
            msg = {"result": False,
                   "message": "min_block missing or too near: refresh your wallet — claims must give "
                              "the network time to propagate before they can be mined"}
            return msg

        # DUPLICATE unbond-withdraw (door-level, user-origin only — same divergence-safety argument as
        # the dividend gate above). The wallet poll re-entered its auto-claim every tick for the whole
        # submit->landing gap and each tick built a fresh withdraw (new nonce -> new txid), so ~20
        # identical claims piled into the pool at once; every copy after the winner died "no pending
        # unbond to withdraw" and the wallet displayed the pile as phantom incoming credit (+200 shown
        # for a 10-coin unbond, 2026-08-19). An account has ONE pending-unbond slot, so one in-flight
        # claim per sender is complete — a second copy can never release more coins. Gossip is exempt:
        # already-admitted duplicates must keep relaying identically on every node.
        elif (user_origin and transaction.get("recipient") == "withdraw"
                and any(t.get("recipient") == "withdraw"
                        and t.get("sender") == transaction.get("sender")
                        for t in self.transaction_pool)):
            msg = {"result": False,
                   "message": "unbond withdrawal already pending — the earlier claim is still waiting "
                              "to land"}
            return msg

        # DUPLICATE dividend claim (same class, observed live 2026-08-19 13:26: 19 copies of one
        # exec-nonce from a single wallet pooled at once — the exec node lists a dividend as unclaimed
        # until the claim LANDS, so wallets re-built it every poll tick). The exec nonce identifies the
        # claim; a second copy for the same (sender, nonce) can never pay out. User-origin only, gossip
        # exempt — identical divergence-safety argument as the two gates above.
        elif (user_origin and transaction.get("recipient") == "dividend_withdraw"
                and any(t.get("recipient") == "dividend_withdraw"
                        and t.get("sender") == transaction.get("sender")
                        and (t.get("data") or {}).get("nonce") == (transaction.get("data") or {}).get("nonce")
                        for t in self.transaction_pool)):
            msg = {"result": False,
                   "message": "dividend claim already pending — the earlier claim is still waiting "
                              "to land"}
            return msg

        # COLLECT-BLOB PROPAGATION GUARD (user-origin only): the exact rule the dividend_withdraw gate
        # enforces, for the collect blob that REQUESTS the dividend. Stale cached wallets still build it
        # with no min_block, and it seeded the ONLY two organic forks after the duty class was closed
        # (h79050 #32, h80403 #36 — both "only_ours: blob/collect_dividend"). The match is deliberately
        # narrow — only op == "collect_dividend", which nothing but the wallet's collect path builds —
        # so game/SDK blobs are untouched. Gossip exempt, as always: door refusals cannot diverge pools.
        elif (user_origin and transaction.get("recipient") == "blob"
                and isinstance(transaction.get("data"), dict)
                and transaction["data"].get("op") == "collect_dividend"
                and int(transaction.get("min_block", 0) or 0)
                        < self.latest_block["block_number"] + 4):
            msg = {"result": False,
                   "message": "min_block missing or too near: refresh your wallet — claims must give "
                              "the network time to propagate before they can be mined"}
            return msg

        # DUPLICATE registration (user-origin only): register is EXACT-landing, so under head churn a
        # wallet's registration misses its block and the wallet re-mints — 50 registrations sat pooled
        # at 16:35 (several senders holding 2-3 copies) while exactly one landed in 40 blocks. A second
        # copy from the same sender while one is still in flight can never register anything extra;
        # refuse it at the door (gossip exempt) so the pool carries one attempt per wallet.
        elif (user_origin and transaction.get("recipient") == "register"
                and any(t.get("recipient") == "register"
                        and t.get("sender") == transaction.get("sender")
                        for t in self.transaction_pool)):
            msg = {"result": False,
                   "message": "registration already pending — the earlier attempt is still waiting "
                              "to land"}
            return msg

        # SUPERSEDED single-duty forms (doc/consensus-aggregation.md): attest/commit/reveal stay
        # consensus-valid FOREVER (historical blocks carry them; genesis sync replays them), but NEW
        # ones are refused at the mempool door — every honest validator emits the merged `duty` tx,
        # which is what bounds the per-epoch consensus load to the committee. LOCAL policy, not a
        # validity rule, so replay of old blocks is untouched.
        elif transaction.get("recipient") in ("attest", "commit", "reveal"):
            return {"result": False, "message": "Superseded by the merged `duty` tx"}

        # AT-MOST-ONCE (2026-07): a txid ALREADY MINED (recorded in the on-chain tx-index by an ancestor
        # block) can never re-enter the mempool. A txid hashes the tx content, so an IDENTICAL transaction
        # has an identical txid — reintroducing/replaying the same transaction is impossible at the entry
        # point. Checked EARLY (one indexed read, before the account read + txid recompute): a re-gossiped
        # already-mined tx is the exact flood the old bug produced. A genuine re-send is a NEW tx (fresh
        # nonce -> fresh txid) the wallet builds on the user's action; it is not blocked here.
        elif isinstance(transaction.get("txid"), str) and kv_ops.tx_get(transaction["txid"]) is not None:
            return {"result": False, "message": "Already mined"}

        # OPEN-lane onboarding: register/heartbeat are fee-exempt ENTRY txs — a brand-new zero-coin
        # address has no on-chain account YET (registration is what creates it), so they must bypass
        # the empty-account anti-spam gate. (register is PoW-gated and heartbeat requires registered=1
        # in validate_transaction, so this opens no spam hole.) Spending txs still need a funded account.
        elif transaction.get("recipient") not in ("register", "heartbeat") \
                and not get_account(transaction["sender"], create_on_error=False):
            msg = {"result": False,
                   "message": f"Empty account"}
            return msg

        elif not validate_txid(transaction, logger=self.logger):  # always enforced (compat gate gone)
            msg = {"result": False,
                   "message": f"Invalid txid"}
            return msg
        # NOTE: the old byte-size validate_base_fee gate is removed: get_byte_size is
        # sys.getsizeof(repr(...)) and is non-deterministic, so it is unsafe as a fee rule.
        # The deterministic MIN_TX_FEE floor is enforced in validate_transaction below.

        else:
            # Re-check by TXID, not by deep dict comparison. validate_txid above guarantees
            # txid == hash(content), so identical content means an identical txid — the O(1) set check
            # is exactly as strong as the old `transaction in self.transaction_pool`, which walked the
            # whole pool doing full-dict __eq__ on every genuinely NEW tx (the one case the fast path
            # at the top of this method cannot short-circuit). Kept rather than deleted because another
            # thread can have added this txid since that fast path ran.
            if transaction.get("txid") in self._pool_txid_set():
                # Idempotent: already pooled (e.g. a re-gossiped heartbeat) — a benign success, not an
                # error (matches the "already present" handling clients now expect).
                return {"message": "Already present", "result": True}
            try:
                validate_transaction(transaction=transaction,
                                     logger=self.logger,
                                     block_height=self.latest_block["block_number"])
            except Exception as e:
                msg = {"result": False,
                       "message": f"Could not merge remote transaction: {e}"}
                return msg
            else:
                try:
                    validate_single_spending(transaction_pool=self.transaction_pool, transaction=transaction)

                    # mutation tail under the mempool lock: the membership re-check and the append+sort
                    # must be atomic vs the core loop's drain/production swaps and vs sibling
                    # merge_transaction calls on other threads (double-accept / lost-append races).
                    with self.mempool_lock:
                        if _txid not in self._pool_txid_set():
                            self._transaction_pool.append(transaction)
                            self.pool_gen += 1   # in-place append — bump the content signal by hand

                except Exception as e:
                    msg = f"Remote transaction failed to validate: {e}"
                    self.logger.info(msg)
                    # NO sender-wide purge (2026-09-01). This used to drop EVERY pooled tx of the sender on
                    # an aggregate-spend refusal — but an "Overspending balance" here is routinely
                    # branch/timing skew (a dividend credit or funding that one node has applied and
                    # another has not yet), not a double-spend attempt; purging made THIS node's pool
                    # diverge from every peer that kept the sender's other txs, and deterministic
                    # production turned that into the 08-31 splits. Double-spends cannot land anyway:
                    # verify_block enforces whole-block spending, and _candidate_pool keeps our own
                    # candidate within balance. Refuse just this tx; it cools briefly and is re-tried.
                    return {"message": msg,
                            "result": False}

            return {"message": "Success", "result": True}

    def merge_transactions(self, transactions, user_origin=False) -> None:
        """Merge a whole remote batch one tx at a time through merge_transaction, which contains its
        own failures — so a single malformed/invalid entry from a malicious peer can never abort the
        rest of the batch. Per-tx results are deliberately discarded (gossip is best-effort)."""
        for transaction in transactions:
            self.merge_transaction(transaction, user_origin)

    def enqueue_gossip(self, transaction, exclude_ip=None) -> None:
        """Queue a NEWLY-accepted tx for push-fan-out to peers (nado._gossip_worker), skipping the
        peer it came from (exclude_ip) so it never echoes straight back. Best-effort: a full queue
        (sustained flood) drops the overflow — the txid-diff pull reconcile still converges it. Only
        the caller's should_gossip(result) gate decides WHAT reaches here, so this never re-floods a
        dup (see ops/gossip.py)."""
        try:
            self.gossip_queue.put_nowait((transaction, exclude_ip))
        except queue.Full:
            pass

    def purge_txs_of_sender(self, sender) -> None:
        """remove all transactions of sender to prevent possible double spending attempt"""
        """of sender sending different txs to different nodes both exhausting balance"""
        # AUDIT FIX: was `for tx in pool: pool.remove(tx)` — removing while iterating shifts the
        # index and SKIPS the element after every hit, so adjacent same-sender txs (the exact
        # double-spend shape this guard exists for) half-survived the purge. Rebuild the pool under
        # the mempool lock instead.
        with self.mempool_lock:
            self.transaction_pool = [t for t in self.transaction_pool if t["sender"] != sender]