#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Eddycoin Node - EXPERIMENTAL / FOR PLAY - test coin only, no real money
=======================================================================
A full Eddycoin node you can run on (almost) any computer: Windows, Linux, macOS, Raspberry Pi.
Python 3.8+ standard library only (tkinter for the window; --headless needs nothing else).

What it does
  * downloads the whole E/D/Y chain from the seed (default: the live testnet server) and/or other nodes,
  * re-validates EVERYTHING itself with the same rules as the server: proof-of-work (double SHA-256),
    difficulty retargeting, Y checkpoints (id = SHA-256(DOMAIN|prev|Y state root|E evidence|D evidence)),
    emission schedule + 900M cap, ECDSA P-256 signatures, UTXO balances, no double spends,
  * keeps syncing, passes new blocks and transactions on to other nodes (plain HTTP),
  * offers the same API as the server, so the CPU/GPU miners and the wallet can use it:
      GET /api/work?chain=E|D   POST /api/submit/e  POST /api/submit/d   GET /api/wallet?addr=  (/api/wallet/<addr>)
      POST /api/tx   GET /api/state[?addr=]   GET /api/verify   GET /api/export   GET /api/block/<E|D|Y>/<height>
      GET /api/peers   (node-to-node: /api/node/info, /api/node/segment, POST /api/node/push, POST /api/node/hello)
  * fork rule: the valid chain with the most proof-of-work wins (sum of 2^bits of all E and D blocks).
  * while a seed (the live server) is reachable and has the best chain, the node FOLLOWS it: miners' blocks and
    transactions are checked locally and passed on to the seed. If no seed is reachable for --seed-timeout seconds
    (default 60), the node runs INDEPENDENTLY: it accepts blocks itself, finalizes checkpoints and continues the chain
    together with the other nodes it knows. That is the "server switched off" case.

Usage
  python3 eddycoin_node.py                      window (GUI)
  python3 eddycoin_node.py --headless           no window (Linux / Raspberry Pi / macOS / servers)
  python3 eddycoin_node.py --headless --listen 0.0.0.0 --port 8770 --peer http://friend.example.org:8770
Then point a miner or the wallet at http://127.0.0.1:8770 (field "Server").
No fixed IP needed. Only if you want OTHER people to connect to YOUR node: forward the port on your router and
(optionally) use a free dynamic-DNS name, then start with --listen 0.0.0.0 --public-url http://yourname.example:8770
"""
import argparse, collections, copy, hashlib, json, math, os, queue, random, re, socket, ssl, sys, threading, time, urllib.error, urllib.parse, urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

NODE_VERSION = "eddycoin-node/1.2"
NETWORK = "eddycoin-testnet-sandbox"
DEFAULT_SEED = "https://eddycoin-demo.pages.dev"
DEFAULT_PORT = 8770
ZERO = "0" * 64
DEC, COIN = 8, 100000000
MAX_SUPPLY = 900000000 * COIN
EM_R0, EM_EPOCH, EM_NUM, EM_DEN = 14985005307, 10000, 998335, 1000000
E_SPLIT_BPS = 5000
MIN_FEE = 1000000
MIN_BITS = 8
D_CP, D_CAND, D_STATE, D_TXROOT, D_TX, D_SIG, D_ADDR = ("EDDY-CHECKPOINT-v1", "EDDY-Y-CANDIDATE|v1", "EDDY-Y-STATE|v1",
                                                         "EDDY-TXROOT|v1", "EDDY-TX|v1", "EDDY-TX-SIG|v1", "EDDY-ADDR|v1")
# live testnet parameters (eddy-core.js LIVE)
TARGET_MS, RETARGET_EVERY, MAX_STEP, MAX_SKEW_MS = 30000, 10, 2, 120000
CLAMP = {"E": (18, 34), "D": (20, 40)}
GENESIS_BITS = {"E": 28, "D": 33}
CAND_MAX_TXS = 40
# ---- difficulty/timestamp fix (soft activation at Y checkpoint height FORK_H; == eddy-core.js FORK) ----
# from FORK_H: retarget from arrival times (recv - time the block's Y candidate opened), block time within +-30 s of arrival and after
# the median of the last 11 block times, checkpoint time = max(E recv, D recv). Blocks/checkpoints below FORK_H keep the old rules.
FORK_H = 300
FORK_SKEW_MS, MTP_N, MAX_FUTURE_MS = 120000, 11, 120000  # skew = old +-120 s so old released miners keep working
if os.environ.get("EDDY_TEST_FORK_H"):    # private local test networks only
    FORK_H = int(os.environ["EDDY_TEST_FORK_H"])
if os.environ.get("EDDY_TEST_CLAMP_LO"):  # e.g. "8,8" (tests: fast mining with retargets)
    _lo = [int(x) for x in os.environ["EDDY_TEST_CLAMP_LO"].split(",")]; CLAMP = {"E": (_lo[0], CLAMP["E"][1]), "D": (_lo[1], CLAMP["D"][1])}
if os.environ.get("EDDY_TEST_CLAMP_HI"):  # e.g. "14,14" (tests: keep pure-Python test mining fast)
    _hi = [int(x) for x in os.environ["EDDY_TEST_CLAMP_HI"].split(",")]; CLAMP = {"E": (CLAMP["E"][0], _hi[0]), "D": (CLAMP["D"][0], _hi[1])}

# ---- cross-checking between nodes (node POLICY only - block / checkpoint validity rules are unchanged) ----
# every XC_INTERVAL s a node asks the seed + a sample of peers for their checkpoint ids (tip, tip-1..tip-10, every 100th);
# a valid fork that would replace checkpoints we hold is adopted only if it has more work AND the seed or >= 2 independent
# peers incl. the sender (and more than hold ours) have it - one peer alone can never flip a node. Invalid data lowers the sender's score
# (BAN_SCORE -> evicted + banned BAN_S s). Mismatches are logged, listed at GET /api/node/mismatches and gossiped
# (POST /api/node/report, informational only - a report never makes a node reject anything; every node checks itself).
XC_INTERVAL, XC_SAMPLE, XC_RECENT, XC_EVERY = 60, 6, 10, 100
SCORE_INVALID, SCORE_CONFLICT, SCORE_GOOD, SCORE_MAX, CONFLICT_FLOOR, BAN_SCORE, BAN_S = -25, -5, 1, 20, -40, -50, 3600
if os.environ.get("EDDY_TEST_XC_INTERVAL"):  # tests only
    XC_INTERVAL = int(os.environ["EDDY_TEST_XC_INTERVAL"])


def indep_key(url):
    """peers on the same host count as ONE independent voice (loopback: per port, for local test networks)"""
    try:
        u = urllib.parse.urlsplit(url); host = (u.hostname or "").lower()
    except Exception:
        return url
    return url if host in ("127.0.0.1", "localhost", "::1") else host


def fork_active(h):
    return isinstance(h, int) and h >= FORK_H


def mtp(arr, n=MTP_N):
    ts = sorted(b["time"] for b in arr[-n:])
    return ts[len(ts) // 2] if ts else None
HEX64 = re.compile(r"^[0-9a-f]{64}$")
ADDR_RE = re.compile(r"^eddy1[0-9a-f]{40}$")
AMT_RE = re.compile(r"^(0|[1-9][0-9]{0,16})$")
BLOCK_KEYS = ("chain", "height", "prev", "y_candidate", "miner", "time", "bits", "nonce", "alg", "hash")


def now_ms():
    return int(time.time() * 1000)


# ======================================================================================================
# protocol functions - line-by-line mirror of src/eddy-core.js (same as eddycoin_ychain_verifier.py)
# ======================================================================================================
def canon(o):
    return json.dumps(o, sort_keys=True, separators=(",", ":"), ensure_ascii=False)


def sha(s):
    return hashlib.sha256(s.encode("utf-8")).hexdigest()


def is_int(n):
    return isinstance(n, int) and not isinstance(n, bool) and abs(n) <= 2 ** 53 - 1


def is_amt(x):
    return isinstance(x, str) and bool(AMT_RE.match(x)) and int(x) <= MAX_SUPPLY


def amt_int(x):
    if isinstance(x, bool):
        raise ValueError("bad amount")
    if isinstance(x, int):
        return x
    if x is None or x == "":
        return 0
    if isinstance(x, str) and re.match(r"^-?[0-9]{1,24}$", x):
        return int(x)
    raise ValueError("bad amount")


def is_addr(a):
    return isinstance(a, str) and bool(ADDR_RE.match(a))


def lzb(h):
    n = 0
    for c in h:
        v = int(c, 16)
        if v == 0:
            n += 4
            continue
        n += 4 - v.bit_length()
        break
    return n


def fmt(m, dec=DEC):
    m = amt_int(m)
    s = str(abs(m)).rjust(dec + 1, "0")
    return ("-" if m < 0 else "") + "{:,}".format(int(s[:-dec])) + "." + s[-dec:]


_EM = None


def _em_table():
    global _EM
    if _EM is None:
        r, c, x, cum = [], [], EM_R0, 0
        while x > 0 and cum < MAX_SUPPLY:
            r.append(x); c.append(cum); cum += x * EM_EPOCH; x = x * EM_NUM // EM_DEN
        _EM = (r, c)
    return _EM


def emission_cum(h):
    if h <= 0:
        return 0
    r, c = _em_table()
    k = (h - 1) // EM_EPOCH
    if k >= len(r):
        return MAX_SUPPLY
    return min(MAX_SUPPLY, c[k] + (h - k * EM_EPOCH) * r[k])


def emission_reward(h):
    return 0 if h < 1 else emission_cum(h) - emission_cum(h - 1)


def next_reduction(h):
    k = max(0, h - 1) // EM_EPOCH
    at = (k + 1) * EM_EPOCH + 1
    return at, emission_reward(at)


def subsidy_at(h, issued):
    return max(0, min(emission_reward(h), MAX_SUPPLY - amt_int(issued)))


def reward_split(sub, fees):
    sub, fees = amt_int(sub), amt_int(fees)
    d = sub // 2 + fees // 2
    return sub + fees - d, d


# ---- ECDSA P-256 verification (pure Python, Jacobian coordinates) ----
_P = 0xffffffff00000001000000000000000000000000ffffffffffffffffffffffff
_A = _P - 3
_B = 0x5ac635d8aa3a93e7b3ebbd55769886bc651d06b0cc53b0f63bce3c3e27d2604b
_N = 0xffffffff00000000ffffffffffffffffbce6faada7179e84f3b9cac2fc632551
_G = (0x6b17d1f2e12c4247f8bce6e563a440f277037d812deb33a0f4a13945d898c296,
      0x4fe342e2fe1a7f9b8ee7eb4a7c0f9e162bce33576b315ececbb6406837bf51f5)


def _jdbl(p):
    X, Y, Z = p
    if Y == 0 or Z == 0:
        return (0, 1, 0)
    YY = Y * Y % _P; S = 4 * X * YY % _P; ZZ = Z * Z % _P
    M = 3 * (X - ZZ) * (X + ZZ) % _P
    X3 = (M * M - 2 * S) % _P
    return (X3, (M * (S - X3) - 8 * YY * YY) % _P, 2 * Y * Z % _P)


def _jadd(p, q):
    if p[2] == 0:
        return q
    if q[2] == 0:
        return p
    Z1Z1 = p[2] * p[2] % _P; Z2Z2 = q[2] * q[2] % _P
    U1 = p[0] * Z2Z2 % _P; U2 = q[0] * Z1Z1 % _P
    S1 = p[1] * q[2] * Z2Z2 % _P; S2 = q[1] * p[2] * Z1Z1 % _P
    if U1 == U2:
        return _jdbl(p) if S1 == S2 else (0, 1, 0)
    H = (U2 - U1) % _P; R = (S2 - S1) % _P; HH = H * H % _P; HHH = H * HH % _P; V = U1 * HH % _P
    X3 = (R * R - HHH - 2 * V) % _P
    return (X3, (R * (V - X3) - S1 * HHH) % _P, H * p[2] * q[2] % _P)


def _jmul(k, p):
    r = (0, 1, 0); q = (p[0], p[1], 1)
    while k:
        if k & 1:
            r = _jadd(r, q)
        q = _jdbl(q); k >>= 1
    return r


def ecdsa_verify(pub_hex, sig_hex, msg):
    try:
        pb = bytes.fromhex(pub_hex); sb = bytes.fromhex(sig_hex)
        if len(pb) != 65 or pb[0] != 4 or len(sb) != 64:
            return False
        Q = (int.from_bytes(pb[1:33], "big"), int.from_bytes(pb[33:], "big"))
        if not (0 <= Q[0] < _P and 0 <= Q[1] < _P) or (Q[1] * Q[1] - (Q[0] ** 3 + _A * Q[0] + _B)) % _P:
            return False
        r, s = int.from_bytes(sb[:32], "big"), int.from_bytes(sb[32:], "big")
        if not (1 <= r < _N and 1 <= s < _N):
            return False
        e = int.from_bytes(hashlib.sha256(msg.encode("utf-8")).digest(), "big")
        w = pow(s, -1, _N)
        X = _jadd(_jmul(e * w % _N, _G), _jmul(r * w % _N, Q))
        if X[2] == 0:
            return False
        x = X[0] * pow(X[2] * X[2] % _P, -1, _P) % _P
        return x % _N == r
    except Exception:
        return False


def block_header(b):
    return "EDDY-%s-BLOCK|v1|%s|%s|%s|%s|%s|%s|%s" % (b["chain"], b["height"], b["prev"], b["y_candidate"], b["miner"], b["time"], b["bits"], b["nonce"])


def block_hash(b):
    if b.get("alg") == "sha256d":
        return hashlib.sha256(hashlib.sha256(block_header(b).encode("utf-8")).digest()).hexdigest()
    return sha(block_header(b))


def check_block(b, chain, min_bits=MIN_BITS):
    try:
        if b.get("chain") != chain:
            return "wrong chain"
        if not all(is_int(b.get(k)) for k in ("height", "nonce", "bits", "time")) or b["height"] < 0 or b["nonce"] < 0:
            return "bad fields"
        if not (b["prev"] == ZERO or HEX64.match(b["prev"])) or not HEX64.match(b["y_candidate"]) or not HEX64.match(b["hash"]):
            return "bad hash fields"
        if not isinstance(b["miner"], str) or len(b["miner"]) > 64:
            return "bad miner"
        if b.get("alg") not in (None, "sha256d"):
            return "unknown hash algorithm"
        if b["bits"] < min_bits or b["bits"] > 40:
            return "difficulty %d below minimum %d" % (b["bits"], min_bits)
        if block_hash(b) != b["hash"]:
            return "hash mismatch"
        if lzb(b["hash"]) < b["bits"]:
            return "hash does not meet target"
    except Exception as ex:
        return "malformed block (%s)" % ex
    return None


def address_of(pub):
    return "eddy1" + sha(D_ADDR + "|" + pub)[:40]


def txid_of(t):
    # eddy-core.js txBody(): {v,type,inputs,outputs,fee,time,memo,pubkey}; canon() drops undefined keys
    return sha(D_TX + "|" + canon({k: t[k] for k in ("v", "type", "inputs", "outputs", "fee", "time", "memo", "pubkey") if k in t}))


def validate_transfer(utxos, t, check_sig=True):
    try:
        if not isinstance(t, dict) or t.get("v") != 2 or t.get("type") != "transfer":
            return "unsupported tx version/type (v2 = string amounts)"
        ins, outs = t.get("inputs"), t.get("outputs")
        if not isinstance(ins, list) or not isinstance(outs, list) or not 1 <= len(ins) <= 50 or not 1 <= len(outs) <= 4:
            return "bad input/output count"
        if not isinstance(t.get("memo"), str) or len(t["memo"]) > 80 or not is_int(t.get("time")) or not re.match(r"^04[0-9a-f]{128}$", str(t.get("pubkey"))):
            return "bad serialization"
        if txid_of(t) != t.get("txid"):
            return "txid mismatch"
        if check_sig and not ecdsa_verify(t["pubkey"], str(t.get("sig", "")), D_SIG + "|" + t["txid"]):
            return "invalid signature"
        owner = address_of(t["pubkey"])
        seen, sin, sout = set(), 0, 0
        for i in ins:
            k = "%s:%s" % (i["txid"], i["vout"])
            if k in seen:
                return "duplicate input"
            seen.add(k)
            u = utxos.get(k)
            if not u:
                return "input missing or already spent (double spend)"
            if u["address"] != owner:
                return "input not owned by signer"
            sin += amt_int(u["amount"])
        for o in outs:
            if not is_amt(o.get("amount")) or int(o["amount"]) <= 0:
                return "output amount must be a positive decimal string"
            if not is_addr(o.get("address")):
                return "bad output address"
            sout += int(o["amount"])
        if not is_amt(t.get("fee")) or int(t["fee"]) < MIN_FEE:
            return "fee below minimum (or not a decimal string)"
        if sin != sout + int(t["fee"]):
            return "value not conserved (in %d != out %d + fee %s)" % (sin, sout, t["fee"])
    except Exception as ex:
        return "malformed transaction (%s)" % ex
    return None


def apply_tx(utxos, t):
    for i in t["inputs"]:
        utxos.pop("%s:%s" % (i["txid"], i["vout"]), None)
    for n, o in enumerate(t["outputs"]):
        utxos["%s:%d" % (t["txid"], n)] = {"address": o["address"], "amount": str(amt_int(o["amount"]))}


def tx_root(txids):
    return sha(D_TXROOT + "|" + ",".join(txids))


def candidate_id(prev, txids, height):
    return sha(D_CAND + "|" + prev + "|" + tx_root(txids) + "|" + str(height))


def state_root(height, issued, utxos):
    lst = ["%s:%s:%s" % (k, utxos[k]["address"], utxos[k]["amount"]) for k in sorted(utxos)]
    return sha(D_STATE + "|" + canon({"height": height, "issued": str(amt_int(issued)), "utxos": lst}))


def checkpoint_id(prev, sroot, e, d):
    return sha(D_CP + "|" + prev + "|" + sroot + "|" + e + "|" + d)


def make_reward(outputs, t, memo):
    tx = {"v": 2, "type": "reward", "inputs": [], "outputs": [{"address": o["address"], "amount": str(amt_int(o["amount"]))} for o in outputs if amt_int(o["amount"]) > 0],
          "fee": "0", "time": t, "memo": memo, "pubkey": "", "sig": ""}
    tx["txid"] = txid_of(tx)
    return tx


def genesis_checkpoint(t=0):
    sr = state_root(0, 0, {})
    cp = {"height": 0, "prev": ZERO, "y_candidate": ZERO, "e_evidence": "GENESIS", "d_evidence": "GENESIS", "e_to": "", "d_to": "", "credit": None,
          "txs": [], "tx_root": tx_root([]), "state_root": sr, "issued": "0", "subsidy": "0", "fees": "0", "time": t}
    cp["id"] = checkpoint_id(ZERO, sr, "GENESIS", "GENESIS")
    return cp


GENESIS_ID = genesis_checkpoint()["id"]


def js_round(x):
    return int(math.floor(x + 0.5))


class Invalid(Exception):
    pass


# ======================================================================================================
# Ledger: the node's own copy of the three chains, built ONLY through validated steps
# ======================================================================================================
class Ledger:
    def __init__(self, genesis_bits=None, genesis_time=0):
        self.gbits = dict(genesis_bits or GENESIS_BITS)
        self.Y = [genesis_checkpoint(genesis_time)]
        self.E, self.D = [], []
        self.fin = {}            # finalized UTXO set (Y state)
        self.issued = 0
        self.bits = dict(self.gbits)
        self.retargets = []
        self.cand = None         # open Y candidate {height, prev, txids, txs, id, e, d, e_miner, d_miner}
        self.mempool = []
        self.utxos = {}          # pending view = fin + candidate txs + mempool
        self.ytx = {}            # finalized txid -> checkpoint height
        self.registry = []       # local log of txs this node accepted (wallet "sent" history)
        self.work = 0

    def tip(self, ch):
        arr = getattr(self, ch)
        return arr[-1] if arr else {"height": -1, "hash": ZERO}

    def snapshot(self):
        return (len(self.Y), len(self.E), len(self.D), dict(self.fin), self.issued, dict(self.bits), list(self.retargets),
                copy.deepcopy(self.cand), list(self.mempool), dict(self.utxos), self.work)

    def restore(self, s):
        ny, ne, nd, self.fin, self.issued, self.bits, self.retargets, self.cand, self.mempool, self.utxos, self.work = s
        for cp in self.Y[ny:]:
            for t in cp["txs"]:
                self.ytx.pop(t["txid"], None)
        del self.Y[ny:]; del self.E[ne:]; del self.D[nd:]

    def retarget(self, ch):  # == eddy-core.js retarget()
        arr = getattr(self, ch); n = RETARGET_EVERY
        if len(arr) < n or (arr[-1]["height"] + 1) % n != 0:
            return None
        last = arr[-n:]; act = fork_active(arr[-1]["height"] + 1)
        # fix 1: solve = arrival (recv) - time the block's Y candidate opened (time of Y checkpoint #height); miner time is ignored
        solve = [max(1000, b["recv"] - self.Y[b["height"]]["time"]) if act and is_int(b.get("recv")) and b["height"] < len(self.Y)
                 else max(1000, (b.get("recv") or b["time"]) - b["time"]) for b in last]
        avg = sum(solve) / n
        step = js_round(math.log2(TARGET_MS / avg)); step = max(-MAX_STEP, min(MAX_STEP, step))
        lo, hi = CLAMP[ch]; old = self.bits[ch]; self.bits[ch] = max(lo, min(hi, old + step))
        self.retargets.append({"chain": ch, "at": arr[-1]["height"], "avg_ms": js_round(avg), "from": old, "to": self.bits[ch], "time": arr[-1].get("recv") or arr[-1]["time"]})
        if len(self.retargets) > 50:
            self.retargets.pop(0)
        return {"from": old, "to": self.bits[ch], "avg_ms": js_round(avg)}

    def rebuild_pending(self, extra=()):
        """pending view = fin + candidate txs + mempool; re-validates the mempool (drops invalid / already included txs)"""
        u = dict(self.fin)
        if self.cand:
            for t in self.cand["txs"]:
                apply_tx(u, t)
        incand = set(self.cand["txids"]) if self.cand else set()
        keep, seen = [], set()
        for t in list(extra) + list(self.mempool):
            tid = t.get("txid") if isinstance(t, dict) else None
            if not tid or tid in seen or tid in incand or tid in self.ytx:
                continue
            seen.add(tid)
            if validate_transfer(u, t) is None:
                apply_tx(u, t); keep.append(t)
        self.mempool = keep; self.utxos = u

    def open_candidate(self):  # == eddy-core.js openCandidate()
        if self.cand:
            return self.cand
        tip = self.Y[-1]; txs = self.mempool[:CAND_MAX_TXS]; self.mempool = self.mempool[CAND_MAX_TXS:]
        txids = [t["txid"] for t in txs]
        self.cand = {"height": tip["height"] + 1, "prev": tip["id"], "txids": txids, "txs": txs, "id": candidate_id(tip["id"], txids, tip["height"] + 1),
                     "e": None, "d": None, "e_miner": None, "d_miner": None}
        return self.cand

    def set_candidate(self, c, check_sig=True):
        """adopt a candidate from a peer/seed (only while our own candidate has no evidence)"""
        if not isinstance(c, dict):
            raise Invalid("bad candidate")
        tip = self.Y[-1]; txs = c.get("txs") or []
        if c.get("prev") != tip["id"] or c.get("height") != tip["height"] + 1:
            raise Invalid("candidate does not extend our Y tip")
        if not isinstance(txs, list) or len(txs) > CAND_MAX_TXS or not all(isinstance(t, dict) for t in txs):
            raise Invalid("bad candidate tx list")
        txids = [t.get("txid") for t in txs]
        if list(c.get("txids") or []) != txids:
            raise Invalid("candidate txids do not match its txs")
        cid = candidate_id(tip["id"], txids, tip["height"] + 1)
        if c.get("id") not in (None, cid):
            raise Invalid("candidate id mismatch")
        if self.cand and self.cand["id"] == cid:
            return self.cand
        if self.cand and (self.cand["e"] or self.cand["d"]):
            raise Invalid("our candidate already has evidence")
        u = dict(self.fin)
        for t in txs:
            why = validate_transfer(u, t, check_sig)
            if why:
                raise Invalid("candidate tx %s: %s" % (str(t.get("txid"))[:12], why))
            apply_tx(u, t)
        old = self.cand["txs"] if self.cand else []
        self.cand = {"height": tip["height"] + 1, "prev": tip["id"], "txids": txids, "txs": txs, "id": cid, "e": None, "d": None, "e_miner": None, "d_miner": None}
        self.rebuild_pending(extra=old)
        return self.cand

    def work_template(self, ch):  # == eddy-core.js liveWork()
        self.open_candidate()
        tip = self.tip(ch); other = "D" if ch == "E" else "E"; waiting = bool(self.cand["e"] if ch == "E" else self.cand["d"])
        msg = ("Y candidate #%d already has %s evidence - waiting for a %s block" % (self.cand["height"], ch, "D-Chain (GPU)" if other == "D" else "E-Chain (CPU)")) if waiting \
            else "mine %s block #%d for Y candidate #%d" % (ch, tip["height"] + 1, self.cand["height"])
        return {"network": NETWORK, "chain": ch, "height": tip["height"] + 1, "prev": tip["hash"], "y_candidate": self.cand["id"], "y_height": self.cand["height"],
                "bits": self.bits[ch], "alg": "sha256d", "waiting": waiting, "waiting_for": other if waiting else None, "message": msg,
                **({"max_time_age_ms": 12000, "time_skew_ms": FORK_SKEW_MS} if fork_active(self.cand["height"]) else {}),
                "header_format": "EDDY-%s-BLOCK|v1|height|prev|y_candidate|miner|time|bits|nonce" % ch, "hash": "sha256d = SHA-256(SHA-256(header))", "server_time": now_ms()}

    # ---------- E/D block = evidence for the open candidate ----------
    def check_evidence(self, b, mode, now=None, skip_time=False):
        """mode 'local': submitted by a miner to this node (same checks as the server's submitBlock);
        mode 'peer': block another node / the seed already accepted (carries its receive time 'recv')"""
        now = now or now_ms()
        if not isinstance(b, dict):
            return "block must be an object"
        ch = b.get("chain")
        if ch not in ("E", "D"):
            return "chain must be E or D"
        if b.get("alg") != "sha256d":
            return "alg must be sha256d"
        if not is_addr(b.get("miner")):
            return "miner must be a wallet address (eddy1 + 40 hex)"
        allowed = BLOCK_KEYS + (("recv",) if mode == "peer" else ())
        for k in b:
            if k not in allowed:
                return "unknown field " + k
        if not self.cand:
            return "no open candidate"
        if b.get("y_candidate") != self.cand["id"]:
            return "stale work: block commits to an old Y candidate (fetch new work)"
        if self.cand["e" if ch == "E" else "d"]:
            return "candidate already has %s evidence - waiting for %s block" % (ch, "D-Chain (GPU)" if ch == "E" else "E-Chain (CPU)")
        if b.get("bits") != self.bits[ch]:
            return "wrong difficulty: %s bits, required %d" % (b.get("bits"), self.bits[ch])
        act = fork_active(self.cand["height"]); skew = FORK_SKEW_MS if act else MAX_SKEW_MS
        if not is_int(b.get("time")):
            return "timestamp outside +-%d s of server time" % (skew // 1000)
        if mode == "local":
            if not skip_time and abs(now - b["time"]) > skew:
                return "timestamp outside +-%d s of server time (check the PC clock / update the miner)" % (skew // 1000)
        else:
            r = b.get("recv")
            if not is_int(r) or abs(r - b["time"]) > skew:
                return "receive time inconsistent with block time"
            if b["time"] > now + MAX_FUTURE_MS or r > now + MAX_FUTURE_MS:
                return "block from the future"
            if act and r < self.Y[-1]["time"]:
                return "receive time before its Y candidate opened"
        arr = getattr(self, ch)
        if arr and b["time"] < arr[-1]["time"] - MAX_SKEW_MS:
            return "timestamp older than the previous block"
        tip = self.tip(ch)
        if b.get("prev") != tip["hash"] or b.get("height") != tip["height"] + 1:
            return "block does not extend the tip"
        return check_block(b, ch)

    def add_evidence(self, b, mode, now=None):
        now = now or now_ms()
        why = self.check_evidence(b, mode, now)
        if why:
            raise Invalid(why)
        ch = b["chain"]; clean = {k: b[k] for k in BLOCK_KEYS}
        clean["recv"] = (max(now, self.Y[-1]["time"]) if fork_active(self.cand["height"]) else now) if mode == "local" else b["recv"]
        getattr(self, ch).append(clean); self.work += 1 << clean["bits"]
        if ch == "E":
            self.cand["e"], self.cand["e_miner"] = clean["hash"], clean["miner"]
        else:
            self.cand["d"], self.cand["d_miner"] = clean["hash"], clean["miner"]
        return self.retarget(ch)

    # ---------- Y checkpoint ----------
    def build_checkpoint(self, t_cp, t_rw):
        """== eddy-core.js finalize(); the times are parameters (the server uses its clock, a node uses max(E time, D time))"""
        c = self.cand; fin = dict(self.fin)
        fees = sum(amt_int(t["fee"]) for t in c["txs"])
        for t in c["txs"]:
            why = validate_transfer(fin, t)
            if why:
                raise Invalid("candidate tx invalid at finalization: " + why)
            apply_tx(fin, t)
        sub = subsidy_at(c["height"], self.issued); se, sd = reward_split(sub, fees)
        rw = make_reward([{"address": c["e_miner"], "amount": se}, {"address": c["d_miner"], "amount": sd}], t_rw,
                         "checkpoint %d reward (E %d%% / D %d%% + fees)" % (c["height"], E_SPLIT_BPS // 100, 100 - E_SPLIT_BPS // 100))
        if rw["outputs"]:
            apply_tx(fin, rw)
        issued = self.issued + sub
        txs = list(c["txs"]) + ([rw] if rw["outputs"] else [])
        sr = state_root(c["height"], issued, fin)
        cp = {"height": c["height"], "prev": c["prev"], "y_candidate": c["id"], "cand_txids": list(c["txids"]), "e_evidence": c["e"], "d_evidence": c["d"],
              "e_to": c["e_miner"], "d_to": c["d_miner"], "credit": {"E": None, "D": None}, "txs": txs, "tx_root": tx_root([t["txid"] for t in txs]),
              "state_root": sr, "issued": str(issued), "subsidy": str(sub), "fees": str(fees), "time": t_cp}
        cp["id"] = checkpoint_id(cp["prev"], sr, cp["e_evidence"], cp["d_evidence"])
        return cp, fin, issued

    def commit_checkpoint(self, cp, fin, issued):
        if issued > MAX_SUPPLY:
            raise Invalid("issuance exceeds the cap")
        self.Y.append(cp); self.fin = fin; self.issued = issued
        for t in cp["txs"]:
            self.ytx[t["txid"]] = cp["height"]
        for e in self.registry:
            if e.get("txid") in self.ytx and e.get("status") != "finalized":
                e["status"] = "finalized"; e["cp"] = self.ytx[e["txid"]]
        self.cand = None
        self.rebuild_pending()

    def finalize_local(self):
        e = self.E[-1]; d = self.D[-1]
        # deterministic: nodes finalizing the same E+D evidence produce the same checkpoint id (fix 1: arrival times, == the server)
        t = max(e["recv"], d["recv"]) if fork_active(self.cand["height"]) else max(e["time"], d["time"])
        cp, fin, issued = self.build_checkpoint(t, t)
        self.commit_checkpoint(cp, fin, issued)
        return cp

    def accept_checkpoint(self, cp):
        """checkpoint made by the seed / another node: rebuild it from OUR candidate + evidence with the rules and compare every field"""
        c = self.cand
        if not c or not c["e"] or not c["d"]:
            raise Invalid("checkpoint without both E and D evidence")
        if not isinstance(cp, dict) or not is_int(cp.get("time")):
            raise Invalid("malformed checkpoint")
        txs = cp.get("txs") or []
        rw = txs[-1] if txs and isinstance(txs[-1], dict) and txs[-1].get("type") == "reward" else None
        t_rw = rw.get("time") if rw else cp["time"]
        if not is_int(t_rw):
            raise Invalid("malformed reward time")
        if fork_active(c["height"]):
            t_fix = max(self.E[-1]["recv"], self.D[-1]["recv"])
            if cp["time"] != t_fix or t_rw != t_fix:
                raise Invalid("checkpoint %s: time must be the later E/D arrival time" % cp.get("height"))
        exp, fin, issued = self.build_checkpoint(cp["time"], t_rw)
        if canon(exp) != canon(cp):
            diff = [k for k in set(exp) | set(cp) if canon(exp.get(k)) != canon(cp.get(k))]
            raise Invalid("checkpoint %s breaks the rules (fields: %s)" % (cp.get("height"), ", ".join(sorted(diff))))
        self.commit_checkpoint(exp, fin, issued)

    def apply_bundle(self, e, d, cp, check_sig=True):
        """one finalized step from the seed or a peer: E block + D block + Y checkpoint"""
        if not isinstance(cp, dict) or cp.get("height") != len(self.Y) or cp.get("prev") != self.Y[-1]["id"]:
            raise Invalid("checkpoint does not link to our Y tip #%d" % (len(self.Y) - 1))
        txs = cp.get("txs") if isinstance(cp.get("txs"), list) else []
        transfers = [t for t in txs if isinstance(t, dict) and t.get("type") == "transfer"]
        cand = {"height": cp["height"], "prev": cp["prev"], "txids": [t.get("txid") for t in transfers], "txs": transfers, "id": cp.get("y_candidate")}
        if self.cand is None or self.cand["id"] != cand["id"]:
            self.set_candidate(cand, check_sig)
        for ch, b in (("E", e), ("D", d)):
            have = self.cand["e" if ch == "E" else "d"]
            if have:
                if not isinstance(b, dict) or have != b.get("hash"):
                    raise Invalid("%s evidence conflicts with our candidate" % ch)
                continue
            self.add_evidence(b, "peer")
        self.accept_checkpoint(cp)


def ledger_from_parts(Y, E, D, cand=None, mempool=(), genesis_bits=None, check_sig=True):
    """full independent replay from genesis; raises Invalid with the first problem found"""
    if not isinstance(Y, list) or not Y or not isinstance(Y[0], dict) or Y[0].get("id") != GENESIS_ID or \
            canon({k: v for k, v in Y[0].items() if k != "time"}) != canon({k: v for k, v in genesis_checkpoint().items() if k != "time"}):
        raise Invalid("genesis checkpoint is not the Eddycoin fair-launch genesis")
    L = Ledger(genesis_bits, Y[0].get("time", 0) if is_int(Y[0].get("time", 0)) else 0)
    n = len(Y) - 1
    if not isinstance(E, list) or not isinstance(D, list) or len(E) < n or len(D) < n or len(E) > n + 1 or len(D) > n + 1:
        raise Invalid("E/D block counts do not match the checkpoints")
    for h in range(1, len(Y)):
        try:
            L.apply_bundle(E[h - 1], D[h - 1], Y[h], check_sig)
        except Invalid as ex:
            raise Invalid("checkpoint %d: %s" % (h, ex))
        except Exception as ex:
            raise Invalid("checkpoint %d: malformed data (%s)" % (h, ex))
    extraE = E[n] if len(E) > n else None
    extraD = D[n] if len(D) > n else None
    if extraE and extraD:
        raise Invalid("open candidate with both E and D evidence but no checkpoint")
    try:
        if extraE or extraD or cand:
            if not cand:
                raise Invalid("evidence block without its candidate")
            L.set_candidate({k: cand.get(k) for k in ("height", "prev", "txids", "txs", "id")}, check_sig)
            for b in (extraE, extraD):
                if b:
                    L.add_evidence(b, "peer")
    except Invalid as ex:
        raise Invalid("open candidate: %s" % ex)
    except Exception as ex:
        raise Invalid("open candidate: malformed (%s)" % ex)
    L.rebuild_pending(extra=[t for t in (mempool or []) if isinstance(t, dict)])
    return L


# ======================================================================================================
# views (same JSON shapes as the server, so the wallet and the miners work unchanged)
# ======================================================================================================
def address_view(L, addr):
    fin = sum(amt_int(u["amount"]) for u in L.fin.values() if u["address"] == addr)
    pend, utxos = 0, []
    for k, u in L.utxos.items():
        if u["address"] == addr:
            pend += amt_int(u["amount"]); utxos.append({"key": k, "amount": str(amt_int(u["amount"]))})
    got = lambda t: sum(amt_int(o["amount"]) for o in t["outputs"] if o["address"] == addr)
    hist = []
    for i in range(len(L.Y) - 1, -1, -1):
        if len(hist) >= 40:
            break
        for t in L.Y[i]["txs"]:
            g = got(t)
            if g > 0:
                hist.append({"txid": t["txid"], "type": t["type"], "amount": str(g), "status": "finalized", "cp": L.Y[i]["height"], "time": L.Y[i]["time"], "memo": t.get("memo")})
    for t in (L.cand["txs"] if L.cand else []) + L.mempool:
        g = got(t)
        if g > 0:
            hist.insert(0, {"txid": t["txid"], "type": t["type"], "amount": str(g), "status": "candidate" if L.cand and t["txid"] in L.cand["txids"] else "unconfirmed", "cp": None, "time": t["time"]})
    pr, waiting = 0, None; c = L.cand
    if c and (c["e_miner"] == addr or c["d_miner"] == addr):
        se, sd = reward_split(subsidy_at(c["height"], L.issued), sum(amt_int(t["fee"]) for t in c["txs"]))
        pr = (se if c["e_miner"] == addr else 0) + (sd if c["d_miner"] == addr else 0)
        waiting = "E" if not c["e"] else "D" if not c["d"] else None
    blocks = {"E": sum(1 for b in L.E if b["miner"] == addr), "D": sum(1 for b in L.D if b["miner"] == addr)}
    sent = [{"txid": e["txid"], "type": "sent", "amount": "-" + str(e["amount"]), "status": tx_status(L, e["txid"])["status"], "note": e.get("note", ""), "time": e["time"]}
            for e in L.registry if e.get("from") == addr][-20:]
    history = sorted(hist + sent, key=lambda x: -(x.get("time") or 0))[:40]
    return {"address": addr, "eddy": str(fin), "eddyPending": str(pend), "pendingReward": str(pr), "waiting": waiting, "blocks": blocks, "utxos": utxos,
            "history": history, "fiat": {"EUR": "0", "CZK": "0"}}


def tx_status(L, txid):
    if L.cand and txid in L.cand["txids"]:
        return {"status": "candidate", "e": bool(L.cand["e"]), "d": bool(L.cand["d"]), "y": False, "cp": None}
    if any(t["txid"] == txid for t in L.mempool):
        return {"status": "unconfirmed", "e": False, "d": False, "y": False, "cp": None}
    if txid in L.ytx:
        return {"status": "finalized", "e": True, "d": True, "y": True, "cp": L.ytx[txid]}
    return {"status": "unknown", "e": False, "d": False, "y": False, "cp": None}


def export_chain(L, label="eddycoin node"):
    cand = {k: L.cand[k] for k in ("height", "prev", "txids", "txs", "id", "e", "d", "e_miner", "d_miner")} if L.cand else None
    return {"format": "eddycoin-chain-export", "version": 1, "network": NETWORK, "label": label, "exported": now_ms(), "max_supply": str(MAX_SUPPLY),
            "subsidy": str(EM_R0), "e_split_bps": E_SPLIT_BPS, "genesis_faucet": "0", "amounts": "decimal-string", "wallets": [],
            "anchor": {"height": -1, "id": ZERO, "utxos": {}, "issued": "0"}, "anchorE": {"height": -1, "hash": ZERO}, "anchorD": {"height": -1, "hash": ZERO},
            "E": L.E, "D": L.D, "Y": L.Y, "mempool": L.mempool, "cand": cand, "registry": [], "issued": str(L.issued),
            "genesis_bits": L.gbits, "node": NODE_VERSION, "node_registry": L.registry[-200:]}


# ======================================================================================================
# HTTP client
# ======================================================================================================
_SSL = None


def http_json(url, body=None, timeout=12, max_bytes=64 * 1024 * 1024):
    global _SSL
    if _SSL is None:
        _SSL = ssl.create_default_context()
    data = None if body is None else json.dumps(body, separators=(",", ":")).encode("utf-8")
    req = urllib.request.Request(url, data=data, method="GET" if body is None else "POST",
                                 headers={"content-type": "application/json", "accept": "application/json", "user-agent": NODE_VERSION + " (test coin)"})
    try:
        with urllib.request.urlopen(req, timeout=timeout, context=_SSL if url.startswith("https://") else None) as r:
            raw = r.read(max_bytes + 1)
            if len(raw) > max_bytes:
                raise IOError("response too large")
            return r.status, json.loads(raw.decode("utf-8"))
    except urllib.error.HTTPError as e:
        try:
            return e.code, json.loads(e.read(1 << 20).decode("utf-8"))
        except Exception:
            return e.code, {"error": "HTTP %d" % e.code}


URL_RE = re.compile(r"^https?://([A-Za-z0-9.\-]{1,100}|\[[0-9A-Fa-f:.]{2,45}\])(:\d{1,5})?$")  # host or [IPv6]


def norm_url(u):
    u = (u or "").strip().rstrip("/")
    if not u:
        return None
    if "://" not in u:
        u = "http://" + u
    return u if URL_RE.match(u) else None


def lan_ips():
    ips = set()
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM); s.connect(("192.0.2.1", 9)); ips.add(s.getsockname()[0]); s.close()
    except Exception:
        pass
    return sorted(i for i in ips if not i.startswith("127."))


# ======================================================================================================
# Node
# ======================================================================================================
class Throttled(IOError):
    pass


class Peer:
    def __init__(self, url, seed=False):
        self.url, self.seed = url, seed
        self.kind = None          # "node" or "server" (the live server has no /api/node/*)
        self.ok = False; self.last_ok = 0; self.last_try = 0; self.err = ""; self.info = {}
        self.same_tip = False; self.pushed = None; self.behind_since = None
        self.score = 0; self.agree = 0; self.disagree = 0; self.invalid = 0; self.checked = 0  # cross-check bookkeeping


class Node:
    def __init__(self, cfg, log=print):
        self.cfg = cfg; self._log = log
        self.lock = threading.RLock()
        self.data = cfg.data
        os.makedirs(self.data, exist_ok=True)
        self.L = None
        self.peers = {}
        for u in cfg.seeds:
            if norm_url(u):
                self.peers[norm_url(u)] = Peer(norm_url(u), seed=True)
        for u in cfg.peers:
            if norm_url(u) and norm_url(u) not in self.peers:
                self.peers[norm_url(u)] = Peer(norm_url(u))
        self.announced = {}       # public URLs of nodes that announced themselves (url -> last seen)
        self.httpd = None; self.stop_ev = threading.Event()
        self.wake = threading.Event(); self.dirty = False; self.last_save = 0
        self.rl = {}; self.status = "starting"; self.verify_cache = None; self.work_cache = {}
        self.events = []
        self.stats = {"accepted": 0, "rejected": 0, "reorgs": 0, "pushes": 0, "from_peers": 0, "invalid_from_peers": 0}
        self.started = time.time()
        # cross-checking (policy): mismatches seen, quarantined forks, reports from other nodes, bans, network events for the window
        self.mismatches = []; self.mm_seen = {}; self.quarantine = {}; self.reports = []; self.banned = {}; self.push_score = {}
        self.xc = {"time": None, "peers": 0, "agree": 0, "disagree": 0, "heights": 0, "status": "not run yet"}
        self.xc_wake = threading.Event(); self.gossip_q = collections.deque(maxlen=50); self.last_gossip = 0
        self.net_ev = collections.deque(maxlen=400)  # (time, url, kind) - drives the window's network animation

    def log(self, msg):
        line = time.strftime("%H:%M:%S ") + msg
        self.events.append(line); del self.events[:-300]
        try:
            self._log(line)
        except Exception:
            pass

    # ---------- persistence ----------
    def chain_path(self):
        return os.path.join(self.data, "chain.json")

    def load(self):
        p = self.chain_path()
        if os.path.exists(p):
            try:
                with open(p, encoding="utf-8") as f:
                    X = json.load(f)
                if X.get("genesis_bits") and X["genesis_bits"] != self.cfg.genesis_bits:
                    raise Invalid("saved chain uses other genesis bits")
                t = time.time()
                self.L = ledger_from_parts(X["Y"], X["E"], X["D"], X.get("cand"), X.get("mempool") or [], self.cfg.genesis_bits)
                self.L.registry = [e for e in (X.get("node_registry") or []) if isinstance(e, dict)][-200:]
                self.log("loaded saved chain: Y #%d, fully re-validated in %.1f s" % (len(self.L.Y) - 1, time.time() - t))
                return
            except Exception as ex:
                bad = p + ".invalid-%d" % int(time.time())
                try:
                    os.replace(p, bad)
                except Exception:
                    pass
                self.log("saved chain rejected (%s) - kept as %s, starting from genesis" % (ex, os.path.basename(bad)))
        self.L = Ledger(self.cfg.genesis_bits)

    def save(self, force=False):
        if not self.dirty and not force:
            return
        with self.lock:
            s = json.dumps(export_chain(self.L), separators=(",", ":"))
            self.dirty = False
        tmp = self.chain_path() + ".tmp"
        with open(tmp, "w", encoding="utf-8") as f:
            f.write(s); f.flush()
            try:
                os.fsync(f.fileno())
            except Exception:
                pass
        os.replace(tmp, self.chain_path()); self.last_save = time.time()

    # ---------- mode ----------
    def following(self):
        """the seed we follow (reachable recently and on the same chain as we are); None = independent"""
        t = time.time()
        for p in self.peers.values():
            if p.seed and p.same_tip and t - p.last_ok < self.cfg.seed_timeout:
                return p
        return None

    def mode(self):
        f = self.following()
        if f:
            return "following the seed %s" % f.url
        if any(p.seed for p in self.peers.values()) and time.time() - self.started < self.cfg.seed_timeout:
            return "syncing"
        return "independent (continuing the chain with other nodes)" if any(p.ok for p in self.peers.values()) else "independent (alone)"

    def changed(self):
        self.dirty = True; self.verify_cache = None; self.wake.set()

    # ---------- chain segments ----------
    def segment(self, frm, limit=300):
        L = self.L; frm = max(1, min(int(frm), len(L.Y)))
        Y = L.Y[frm:frm + limit]; more = frm + limit < len(L.Y)
        if more:
            E = L.E[frm - 1:frm - 1 + len(Y)]; D = L.D[frm - 1:frm - 1 + len(Y)]; cand = None
        else:
            E = L.E[frm - 1:]; D = L.D[frm - 1:]
            cand = {k: L.cand[k] for k in ("height", "prev", "txids", "txs", "id")} if L.cand else None
        return {"from": frm, "Y": Y, "E": E, "D": D, "cand": cand, "mempool": [] if more else L.mempool[:100], "more": more,
                "tip": len(L.Y) - 1, "work": str(L.work), "genesis": GENESIS_ID, "network": NETWORK}

    def consider(self, seg, prefer=False, source="peer"):
        """validate a chain segment from a seed/peer and adopt it if it is better. returns (changed, need_from, result)"""
        try:
            frm = int(seg.get("from", 1)); Y = seg.get("Y") or []; E = seg.get("E") or []; D = seg.get("D") or []
            cand = seg.get("cand"); mem = [t for t in (seg.get("mempool") or []) if isinstance(t, dict)][:200]
            if seg.get("network", NETWORK) != NETWORK or seg.get("genesis", GENESIS_ID) != GENESIS_ID:
                return False, None, "other network"
            if not (isinstance(Y, list) and isinstance(E, list) and isinstance(D, list)) or frm < 1 or any(not isinstance(x, dict) for x in Y + E + D):
                return False, None, "INVALID: malformed segment"
            if cand is not None and not isinstance(cand, dict):
                return False, None, "INVALID: malformed candidate"
        except Exception:
            return False, None, "INVALID: malformed segment"
        with self.lock:
            L = self.L
            if frm > len(L.Y):
                return False, len(L.Y), "gap"
            k, i = frm, 0
            while i < len(Y) and k < len(L.Y) and Y[i].get("height") == k and Y[i].get("id") == L.Y[k]["id"]:
                k += 1; i += 1
            newY = Y[i:]
            if newY and (newY[0].get("height") != k or newY[0].get("prev") != L.Y[k - 1]["id"]):
                if frm <= 1:
                    return False, None, "INVALID: does not connect to genesis"
                return False, max(1, frm - 100), "fork below segment start"
            if not newY and k < len(L.Y):
                return False, None, "peer is behind"
            tipH = k - 1 + len(newY)

            def blk(arr, h):  # segment arrays start at chain height frm-1
                j = h - (frm - 1)
                return arr[j] if 0 <= j < len(arr) else None
            extraE, extraD = blk(E, tipH), blk(D, tipH)
            pure = (k == len(L.Y)); conflict = False
            if pure and L.cand and (L.cand["e"] or L.cand["d"]):
                pid = newY[0].get("y_candidate") if newY else (cand or {}).get("id")
                if pid is not None and pid != L.cand["id"]:
                    conflict = True
                for have, b in ((L.cand["e"], blk(E, k - 1)), (L.cand["d"], blk(D, k - 1))):
                    if have and b is not None and b.get("hash") != have:
                        conflict = True
            if pure and not conflict and not prefer and not newY and not extraE and not extraD:
                return False, None, "nothing new"
            if pure and not conflict:
                snap = L.snapshot(); before = (len(L.Y), L.work, L.cand and L.cand["id"])
                try:
                    for j, cp in enumerate(newY):
                        h = k + j
                        L.apply_bundle(blk(E, h - 1), blk(D, h - 1), cp)
                    if cand:
                        L.set_candidate(cand)
                    elif extraE or extraD:
                        raise Invalid("evidence block without its candidate")
                    for b in (extraE, extraD):
                        if b is not None and not (L.cand and L.cand["e" if b.get("chain") == "E" else "d"] == b.get("hash")):
                            L.add_evidence(b, "peer")
                    if L.cand and L.cand["e"] and L.cand["d"]:
                        cp = L.finalize_local(); L.open_candidate()
                        self.log("E + D evidence complete -> Y checkpoint #%d finalized" % cp["height"])
                    if mem:
                        L.rebuild_pending(extra=mem)
                except Exception as ex:
                    L.restore(snap); self.stats["invalid_from_peers"] += 1
                    return False, None, "INVALID: %s" % ex
                after = (len(L.Y), L.work, L.cand and L.cand["id"])
                if after != before:
                    if newY:
                        self.log("+%d checkpoint(s) from %s, validated -> Y #%d" % (len(newY), source, len(L.Y) - 1))
                    self.stats["from_peers"] += len(newY); self.changed()
                    return True, None, "extended"
                return False, None, "nothing new"
            # a fork (or a competing open candidate): rebuild the alternative chain from genesis and compare the work
            try:
                altE = L.E[:k - 1] + [blk(E, h) for h in range(k - 1, tipH)] + ([extraE] if extraE else [])
                altD = L.D[:k - 1] + [blk(D, h) for h in range(k - 1, tipH)] + ([extraD] if extraD else [])
                if any(b is None for b in altE + altD):
                    return False, max(1, frm - 100), "segment incomplete"
                alt = ledger_from_parts(L.Y[:k] + newY, altE, altD, cand, mem, self.cfg.genesis_bits)
            except Exception as ex:
                self.stats["invalid_from_peers"] += 1
                return False, None, "INVALID: %s" % ex
            differs = alt.Y[-1]["id"] != L.Y[-1]["id"] or (alt.cand or {}).get("id") != (L.cand or {}).get("id") or alt.work != L.work
            if not (alt.work > L.work or (prefer and alt.work == L.work and differs)):
                return False, None, "less work"
            if k >= len(L.Y) or prefer:  # only the open candidate differs, or the seed: normal fork choice (unchanged)
                self._adopt(L, alt, k, source, "more work" if alt.work > L.work else "equal work, seed preferred")
                return True, None, "reorg"
            # a VALID, heavier chain that would replace checkpoints we hold: policy - never let one peer alone flip us
            fork_h, alt_id, ours_id, alt_tip = k, alt.Y[k]["id"], L.Y[k]["id"], alt.Y[-1]["id"]
            q = self.quarantine.get(alt_tip)
            if q and time.time() - q["checked"] < 30:
                q["seen"] += 1
                return False, None, "quarantined (%s)" % q["detail"]
        ok, detail = self.confirm_fork(fork_h, alt_id, ours_id, source)
        with self.lock:
            if ok and self.L is L and alt.work > L.work:
                self.quarantine.pop(alt_tip, None)
                self._adopt(L, alt, fork_h, source, "more work, %s" % detail)
                return True, None, "reorg"
            new = alt_tip not in self.quarantine
            self.quarantine[alt_tip] = {"tip": alt_tip, "fork_height": fork_h, "tipY": len(alt.Y) - 1, "work": str(alt.work), "our_work": str(L.work),
                                        "ours": ours_id, "theirs": alt_id, "source": source, "detail": detail, "first": now_ms(),
                                        "checked": time.time(), "seen": 1 if new else self.quarantine[alt_tip]["seen"] + 1}
            for old_tip in sorted(self.quarantine, key=lambda t: self.quarantine[t]["checked"])[:-5]:
                self.quarantine.pop(old_tip, None)
        if new:
            self.record_mismatch(source, fork_h, ours_id, alt_id, "conflicting-fork",
                                 "valid heavier fork (to Y #%d) that would replace our checkpoints - NOT adopted, quarantined (%s)" % (len(alt.Y) - 1, detail))
            self.adjust(source, SCORE_CONFLICT, "conflict")
        return False, None, "quarantined (%s)" % detail

    def _adopt(self, L, alt, k, source, why):
        old = (L.cand["txs"] if L.cand else []) + L.mempool + [t for cp in L.Y[k:] for t in cp["txs"] if t.get("type") == "transfer"]
        alt.registry = L.registry
        alt.rebuild_pending(extra=old)
        self.log("REORG to the chain from %s: Y #%d -> #%d (%s)" % (source, len(L.Y) - 1, len(alt.Y) - 1, why))
        self.L = alt; self.stats["reorgs"] += 1; self.changed(); self.net_event(source, "data")

    # ---------- cross-checking (policy, not consensus) ----------
    def net_event(self, url, kind):
        if isinstance(url, str) and url.startswith("http"):
            self.net_ev.append((time.time(), url, kind))

    def is_banned(self, url):
        t = self.banned.get(url)
        if t and time.time() < t:
            return True
        if t:
            self.banned.pop(url, None)
        return False

    def adjust(self, url, delta, why):
        """peer score: invalid data -25 (ban + evict at -50), conflicting valid fork -5 (never below -40: no ban for valid data), agreement +1"""
        p = self.peers.get(url) if isinstance(url, str) else None
        if p is None:
            return
        if why == "conflict":
            p.score = min(p.score, max(CONFLICT_FLOOR, p.score + delta))
        else:
            p.score = max(-100, min(SCORE_MAX, p.score + delta))
        if p.score <= BAN_SCORE and not p.seed:
            self.banned[url] = time.time() + BAN_S; self.peers.pop(url, None)
            self.log("EVICTED peer %s (score %d after %d invalid answers) - banned for %d min" % (url, p.score, p.invalid, BAN_S // 60))
            self.net_event(url, "evicted")

    def on_invalid(self, url, height, why, push_ip=None):
        if push_ip is not None:  # pushes are scored by sender IP (a pushed from_url could be forged)
            sc = self.push_score.get(push_ip, 0) + SCORE_INVALID; self.push_score[push_ip] = sc
            if sc <= BAN_SCORE and push_ip not in ("127.0.0.1", "::1", "?"):  # never ban this computer's own programs
                self.banned["push@" + push_ip] = time.time() + BAN_S
                self.log("pushes from %s refused for %d min (invalid data)" % (push_ip, BAN_S // 60))
        p = self.peers.get(url) if isinstance(url, str) else None
        if p:
            p.invalid += 1
        self.record_mismatch(url if isinstance(url, str) else ("push from %s" % push_ip), height, None, None, "invalid", "invalid data REJECTED: %s" % str(why)[:160])
        if push_ip is None:
            self.adjust(url, SCORE_INVALID, "invalid")

    def record_mismatch(self, peer, height, ours, theirs, kind, verdict, gossip=True):
        key = (peer, height, theirs, kind); t = time.time()
        if t - self.mm_seen.get(key, 0) < 1800:
            return
        self.mm_seen[key] = t
        if len(self.mm_seen) > 2000:
            self.mm_seen.clear()
        e = {"time": now_ms(), "peer": peer, "height": height, "ours": ours, "theirs": theirs, "kind": kind, "verdict": verdict}
        self.mismatches.append(e); del self.mismatches[:-300]
        self.log("MISMATCH %s at Y #%s: ours %s vs theirs %s -> %s" % (peer, height, (ours or "-")[:12], (theirs or "-")[:12], verdict))
        self.net_event(peer, "mismatch")
        if gossip:
            self.gossip_q.append(e); self.xc_wake.set()

    def ckp_ids(self, heights):
        L = self.L; out = {}
        for h in heights:
            if isinstance(h, int) and 0 <= h < len(L.Y):
                out[str(h)] = [L.Y[h]["id"], L.E[h - 1]["hash"] if 1 <= h <= len(L.E) else None, L.D[h - 1]["hash"] if 1 <= h <= len(L.D) else None]
        return out

    def fetch_ids(self, p, heights):
        """height -> checkpoint id as held by peer p (node: GET /api/node/checkpoints; seed server: public GET /api/state + /api/block/Y/<h>)"""
        if p.kind == "node":
            st, r = http_json("%s/api/node/checkpoints?h=%s" % (p.url, ",".join(str(h) for h in heights[:40])), timeout=10, max_bytes=256 * 1024)
            if st != 200 or not isinstance(r, dict) or r.get("genesis") != GENESIS_ID:
                raise IOError("checkpoints: HTTP %d" % st)
            ids = r.get("ids") or {}
            return {int(h): v[0] for h, v in ids.items() if str(h).isdigit() and isinstance(v, list) and v and isinstance(v[0], str)}
        if p.kind == "server":
            st, r = http_json(p.url + "/api/state", timeout=15, max_bytes=8 * 1024 * 1024)
            if st != 200 or not isinstance(r, dict):
                raise IOError("state: HTTP %d" % st)
            out = {y["height"]: y["id"] for y in (r.get("Y") or []) if isinstance(y, dict) and isinstance(y.get("height"), int)}
            extra = [h for h in heights if h not in out and isinstance(h, int)]
            for h in sorted(extra, reverse=True)[:2]:
                st, b = http_json("%s/api/block/Y/%d" % (p.url, h), timeout=10, max_bytes=256 * 1024)
                if st == 200 and isinstance(b, dict) and b.get("height") == h:
                    out[h] = b.get("id")
            return {h: out[h] for h in heights if h in out}
        raise IOError("unknown peer kind")

    def confirm_fork(self, h, alt_id, ours_id, source):
        """does the seed, or a quorum of independent peers (>= 2 and more than hold ours), hold checkpoint alt_id at height h?"""
        src_key = indep_key(source) if isinstance(source, str) and source.startswith("http") else None
        votes_alt, votes_ours = ({src_key} if src_key else set()), set(); seed_says = None  # the sender itself is ONE voice
        cands = [p for p in list(self.peers.values()) if p.ok and p.url != source and p.kind in ("node", "server")]
        seeds = [p for p in cands if p.seed]; others = [p for p in cands if not p.seed]; random.shuffle(others)
        for p in seeds + others[:8]:
            try:
                v = self.fetch_ids(p, [h]).get(h)
            except Exception:
                continue
            self.net_event(p.url, "check")
            key = indep_key(p.url)
            if v is None or key == src_key:
                continue
            if p.seed and seed_says is None:
                seed_says = v
            if v == alt_id:
                votes_alt.add(key)
            elif v == ours_id:
                votes_ours.add(key)
        if seed_says is not None and seed_says == alt_id:
            return True, "confirmed by the seed"
        if len(votes_alt) >= 2 and len(votes_alt) > len(votes_ours):
            return True, "confirmed by %d independent peers" % len(votes_alt)
        return False, "%d independent peer(s) incl. the sender hold it, %d hold ours%s" % (
            len(votes_alt), len(votes_ours), "" if seed_says is None else ", the seed holds %s" % ("ours" if seed_says == ours_id else "another one"))

    def cross_check(self):
        with self.lock:
            tipY = len(self.L.Y) - 1
            hs = [h for h in range(tipY, tipY - XC_RECENT - 1, -1) if h >= 1] + [h for h in range(XC_EVERY, tipY + 1, XC_EVERY)][-20:]
            hs = sorted(set(hs), reverse=True)
            ours = {h: self.L.Y[h]["id"] for h in hs}
        # no extra load on the seed server (Cloudflare Worker): while we follow it our chain IS its chain; otherwise ask it at most every 10 min
        f = self.following(); t = time.time()
        cands = [p for p in list(self.peers.values()) if p.ok and (p.kind == "node" or (p.kind == "server" and p is not f and t - getattr(p, "xc_t", 0) > 600))]
        for p in cands:
            if p.kind == "server":
                p.xc_t = t
        seeds = [p for p in cands if p.seed]; others = [p for p in cands if not p.seed]; random.shuffle(others)
        res = {}
        for p in seeds + others[:XC_SAMPLE]:
            if self.stop_ev.is_set():
                return
            try:
                res[p.url] = self.fetch_ids(p, hs); p.checked += 1; self.net_event(p.url, "check")
            except Exception:
                continue
        agree = disagree = 0
        with self.lock:
            if any(self.L.Y[h]["id"] != ours[h] for h in hs if h < len(self.L.Y)) or len(self.L.Y) <= max(hs or [0]):
                self.xc["status"] = "our chain changed during the check - next round"; return
        for url, ids in res.items():
            diff = sorted(h for h, v in ids.items() if h in ours and v != ours[h])
            p = self.peers.get(url)
            if not diff:
                agree += 1
                if p:
                    p.agree += 1; self.adjust(url, SCORE_GOOD, "agree")
                continue
            disagree += 1; h = diff[0]
            if p:
                p.disagree += 1
            votes = {}
            for u2, ids2 in res.items():
                if h in ids2:
                    votes.setdefault(ids2[h], set()).add(indep_key(u2))
            votes.setdefault(ours[h], set()).add("self")
            seed_v = next((res[s.url].get(h) for s in seeds if s.url in res and h in res[s.url]), None)
            we_ok = (seed_v == ours[h]) if seed_v is not None else len(votes.get(ours[h], ())) >= len(votes.get(ids[h], ()))
            verdict = ("peer holds a different fork (minority%s) - not adopted" % (", the seed agrees with us" if seed_v == ours[h] else "")) if we_ok else \
                      ("OUR node is in the minority at this height - we switch only to a heavier chain confirmed by the seed or >= 2 independent peers")
            self.record_mismatch(url, h, ours[h], ids[h], "checkpoint-id", verdict)
            if we_ok:
                self.adjust(url, -2, "conflict")
            else:
                self.wake.set()
        self.xc = {"time": now_ms(), "peers": len(res), "agree": agree, "disagree": disagree, "heights": len(hs),
                   "status": "%d of %d peers agree on %d recent checkpoints" % (agree, len(res), len(hs)) if res else "no peer answered"}

    def send_reports(self):
        if not self.gossip_q or time.time() - self.last_gossip < 10:
            return
        self.last_gossip = time.time()
        batch = [self.gossip_q.popleft() for _ in range(min(5, len(self.gossip_q)))]
        targets = [p for p in list(self.peers.values()) if p.ok and p.kind == "node"][:8]
        for e in batch:
            body = dict(e, from_url=norm_url(self.cfg.public_url) if self.cfg.public_url else None)
            for p in targets:
                if p.url == e.get("peer"):
                    continue
                try:
                    http_json(p.url + "/api/node/report", body, timeout=6); self.net_event(p.url, "report")
                except Exception:
                    pass

    def api_report(self, body, ip):
        """an informational mismatch notice from another node: stored + shown, triggers our OWN cross-check; never causes a rejection"""
        if not isinstance(body, dict):
            return 400, {"ok": False, "error": "bad body"}
        cl = lambda v, n: v[:n] if isinstance(v, str) else None
        h = body.get("height") if isinstance(body.get("height"), int) else None
        e = {"time": now_ms(), "reporter_ip": ip, "reporter": cl(body.get("from_url"), 120), "peer": cl(body.get("peer"), 120), "height": h,
             "ours": cl(body.get("ours"), 64), "theirs": cl(body.get("theirs"), 64), "kind": cl(body.get("kind"), 30), "verdict": cl(body.get("verdict"), 240)}
        self.reports.append(e); del self.reports[:-200]
        self.log("report from %s: %s at Y #%s (%s) - informational, checking ourselves" % (e["reporter"] or ip, e["peer"], h, e["kind"]))
        self.xc_wake.set()
        return 200, {"ok": True, "note": "informational only - this node verifies everything itself"}

    def api_mismatches(self):
        t = time.time()
        return {"mismatches": self.mismatches[-100:], "quarantine": [{k: v for k, v in q.items() if k != "checked"} for q in self.quarantine.values()],
                "reports": self.reports[-50:], "last_check": self.xc,
                "peers": [{"url": p.url, "seed": p.seed, "ok": p.ok, "score": p.score, "agree": p.agree, "disagree": p.disagree, "invalid": p.invalid}
                          for p in list(self.peers.values())],
                "banned": {u: int(until - t) for u, until in list(self.banned.items()) if until > t}}

    def xc_loop(self):
        last = time.time() - XC_INTERVAL + min(15, XC_INTERVAL)  # first round ~15 s after start (peers connected, chain loaded)
        while not self.stop_ev.is_set():
            forced = self.xc_wake.is_set(); self.xc_wake.clear()
            try:
                self.send_reports()
                if time.time() - last >= XC_INTERVAL or (forced and time.time() - last >= 10):
                    last = time.time(); self.cross_check()
            except Exception as ex:
                self.xc["status"] = "check error: %s" % str(ex)[:100]
            self.xc_wake.wait(2.0)

    # ---------- talking to the seed server and to other nodes ----------
    def seed_work(self, url, ch, max_age=2.0):
        """GET <seed>/api/work with a short cache shared by syncing and the miner proxy (saves the seed's per-IP rate limit)"""
        k = (url, ch); c = self.work_cache.get(k)
        if c and time.time() - c[0] < max_age:
            return c[1], dict(c[2])
        st, w = http_json("%s/api/work?chain=%s" % (url, ch), timeout=10)
        if st == 200:
            self.work_cache[k] = (time.time(), st, w)
        return st, dict(w) if isinstance(w, dict) else w

    def server_segment(self, p):
        """segment from the live server (Cloudflare Worker) using only its public GET API"""
        st1, wE = self.seed_work(p.url, "E"); st2, wD = self.seed_work(p.url, "D")
        if 429 in (st1, st2):
            raise Throttled("seed is rate-limiting this IP - waiting")
        if "y_candidate" not in wE or "y_candidate" not in wD:
            raise IOError("server: %s" % (wE.get("error") or wD.get("error") or "HTTP %d/%d" % (st1, st2)))
        if wE.get("network") != NETWORK:
            raise IOError("server on another network")
        T = wE["y_height"] - 1
        p.info = {"tipY": T, "tips": {"E": wE["height"] - 1, "D": wD["height"] - 1}, "cand_id": wE["y_candidate"]}
        with self.lock:
            L = self.L
            if L.cand and L.cand["id"] == wE["y_candidate"] and L.tip("E")["hash"] == wE["prev"] and L.tip("D")["hash"] == wD["prev"]:
                return None
            # same open candidate, but we hold evidence the seed lacks (accepted while the seed was unreachable): offer it to the seed first
            resub = []
            if L.cand and L.cand["id"] == wE["y_candidate"]:
                if L.cand["e"] and not wE.get("waiting") and L.E and L.E[-1]["hash"] == L.cand["e"]:
                    resub.append(("e", {k: L.E[-1][k] for k in BLOCK_KEYS}))
                if L.cand["d"] and not wD.get("waiting") and L.D and L.D[-1]["hash"] == L.cand["d"]:
                    resub.append(("d", {k: L.D[-1][k] for k in BLOCK_KEYS}))
        for lane, b in resub:
            st, r = http_json("%s/api/submit/%s" % (p.url, lane), b, timeout=20)
            self.log("offered our %s block #%d to the seed: %s" % (lane.upper(), b["height"], "accepted" if r.get("ok") else r.get("reason") or r.get("error")))
        if resub:
            st1, wE = self.seed_work(p.url, "E", 0); st2, wD = self.seed_work(p.url, "D", 0)
            if "y_candidate" not in wE or "y_candidate" not in wD:
                raise Throttled("seed busy")
            T = wE["y_height"] - 1
            p.info = {"tipY": T, "tips": {"E": wE["height"] - 1, "D": wD["height"] - 1}, "cand_id": wE["y_candidate"]}
        with self.lock:
            L = self.L
            ourL = len(L.Y)
            ourprev = L.Y[T]["id"] if T < len(L.Y) else None
        frm = max(1, min(ourL, T + 1) - 1)
        if T - frm > 60:
            return self.export_segment(p)
        Y, E, D = [], [], []
        for h in range(frm, T + 1):
            for ch, arr, hh in (("Y", Y, h), ("E", E, h - 1), ("D", D, h - 1)):
                st, b = http_json("%s/api/block/%s/%d" % (p.url, ch, hh))
                if st != 200:
                    raise IOError("server block %s/%d: HTTP %d" % (ch, hh, st))
                arr.append(b)
        for ch, w, arr in (("E", wE, E), ("D", wD, D)):
            if w.get("waiting"):
                st, b = http_json("%s/api/block/%s/%d" % (p.url, ch, w["height"] - 1))
                if st != 200:
                    raise IOError("server evidence block: HTTP %d" % st)
                arr.append(b)
        prev = Y[-1]["id"] if Y else ourprev
        if prev is None or wE["y_candidate"] != candidate_id(prev, [], T + 1):
            return self.export_segment(p)  # candidate carries transactions (or we are far apart): the export has the tx bodies
        return {"from": frm, "Y": Y, "E": E, "D": D, "cand": {"height": T + 1, "prev": prev, "txids": [], "txs": [], "id": wE["y_candidate"]}, "mempool": []}

    def export_segment(self, p):
        st, X = http_json(p.url + "/api/export", timeout=90)
        if st == 429:
            raise Throttled("seed export rate-limited - retrying later")
        if st != 200 or X.get("format") != "eddycoin-chain-export":
            raise IOError("export failed: HTTP %d" % st)
        if (X.get("anchor") or {}).get("height", -1) >= 0:
            raise IOError("server export is pruned - cannot replay from genesis")
        if not X.get("Y") or X["Y"][0].get("id") != GENESIS_ID:
            raise IOError("server genesis differs")
        cand = X.get("cand")
        return {"from": 1, "Y": X["Y"][1:], "E": X.get("E") or [], "D": X.get("D") or [],
                "cand": {k: cand.get(k) for k in ("height", "prev", "txids", "txs", "id")} if isinstance(cand, dict) else None, "mempool": X.get("mempool") or []}

    def sync_peer(self, p):
        p.last_try = time.time()
        try:
            if p.kind is None:
                st, info = http_json(p.url + "/api/node/info")
                if st == 200 and info.get("node"):
                    p.kind = "node"
                else:
                    st2, w = http_json(p.url + "/api/work?chain=E")
                    if st2 == 200 and w.get("network") == NETWORK:
                        p.kind = "server"
                    else:
                        raise IOError("not an Eddycoin node or server (HTTP %d)" % st)
            if p.kind == "node":
                st, info = http_json(p.url + "/api/node/info")
                if st == 429:
                    raise Throttled("peer is rate-limiting - waiting")
                if st != 200:
                    raise IOError("HTTP %d" % st)
                if info.get("genesis") != GENESIS_ID or info.get("network") != NETWORK:
                    raise IOError("peer is on another network")
                p.info = info
                for u in (info.get("peers") or [])[:20]:
                    self.add_peer(u, quiet=True)
                with self.lock:
                    L = self.L; ourw = L.work; ourtip = L.Y[-1]["id"]
                    ourc = (L.cand["id"], L.cand["e"], L.cand["d"]) if L.cand else None
                theirc = (info.get("cand_id"), info.get("cand_e"), info.get("cand_d")) if info.get("cand_id") else None
                tw = int(info.get("work") or "0")
                same = info.get("tip_id") == ourtip and theirc == ourc
                if not same and (tw > ourw or (p.seed and tw == ourw)):
                    frm = max(1, min(len(self.L.Y), int(info.get("tipY", 0)) + 1) - 2)
                    for _ in range(12):
                        st, seg = http_json("%s/api/node/segment?from=%d" % (p.url, frm), timeout=90)
                        if st != 200:
                            raise IOError("segment: HTTP %d" % st)
                        ch, need, why = self.consider(seg, prefer=p.seed, source=p.url)
                        if need:
                            frm = need; continue
                        if why.startswith("INVALID"):
                            self.log("REJECTED data from %s: %s" % (p.url, why)); self.on_invalid(p.url, seg.get("from") if isinstance(seg, dict) else None, why)
                        elif why == "extended":
                            self.net_event(p.url, "data")
                        if seg.get("more") and why in ("extended", "reorg"):
                            frm = len(self.L.Y) - 1; continue
                        break
                elif tw < ourw:
                    self.push(p)
                if p.seed:
                    with self.lock:
                        p.same_tip = tw >= self.L.work or info.get("tip_id") == self.L.Y[-1]["id"]
            else:
                seg = self.server_segment(p)
                why = "same"
                if seg:
                    ch, need, why = self.consider(seg, prefer=True, source=p.url)
                    if need or why == "fork below segment start":
                        seg = self.export_segment(p); ch, need, why = self.consider(seg, prefer=True, source=p.url)
                    if why.startswith("INVALID"):
                        self.log("REJECTED data from %s: %s" % (p.url, why)); self.on_invalid(p.url, seg.get("from") if isinstance(seg, dict) else None, why)
                    elif why in ("extended", "reorg"):
                        self.net_event(p.url, "data")
                # we follow the seed while it has the best chain (we adopted it or already had it)
                # a seed that is a few seconds behind is normal (lag); only if it stays behind / on a weaker chain for
                # --seed-timeout seconds do we stop following it
                if why in ("same", "extended", "nothing new", "reorg"):
                    p.same_tip = True; p.behind_since = None
                elif p.same_tip:
                    p.behind_since = p.behind_since or time.time()
                    if time.time() - p.behind_since > self.cfg.seed_timeout:
                        p.same_tip = False
                        self.log("the seed %s stayed behind / on a weaker chain (%s) - continuing independently" % (p.url, why))
            if not p.ok:
                self.log("connected to %s (%s)" % (p.url, "seed server" if p.kind == "server" else "seed node" if p.seed else "node"))
                self.net_event(p.url, "connected")
            p.ok = True; p.last_ok = time.time(); p.err = ""
        except Throttled as ex:
            p.last_ok = time.time(); p.err = str(ex)  # alive, just busy: not an outage
        except Exception as ex:
            if p.ok or not p.err:
                self.log("%s unreachable: %s" % (p.url, str(ex)[:120]))
            p.ok = False; p.err = str(ex)[:160]

    def push(self, p):
        """send our newest checkpoints + open candidate to a node; it validates them and adopts them if they have more work"""
        if p.kind != "node":
            return
        with self.lock:
            key = (self.L.Y[-1]["id"], self.L.cand and (self.L.cand["id"], self.L.cand["e"], self.L.cand["d"]))
            if p.pushed == key:
                return
            body = self.segment(max(1, len(self.L.Y) - 3))
        body["from_url"] = norm_url(self.cfg.public_url) if self.cfg.public_url else None
        for _ in range(8):
            try:
                st, r = http_json(p.url + "/api/node/push", body, timeout=60)
            except Exception as ex:
                p.err = "push: %s" % str(ex)[:100]; return
            if st == 200 and r.get("need_from"):
                with self.lock:
                    body = self.segment(int(r["need_from"]))
                body["from_url"] = norm_url(self.cfg.public_url) if self.cfg.public_url else None
                continue
            if st == 200 and str(r.get("result", "")).startswith("INVALID"):
                self.log("peer %s rejected our data: %s" % (p.url, r.get("result")))
            p.pushed = key; self.stats["pushes"] += 1; self.net_event(p.url, "push")
            return

    def add_peer(self, u, quiet=False):
        u = norm_url(u)
        if not u or u in self.peers or len(self.peers) >= 40 or self.is_banned(u):
            return
        if self.cfg.public_url and u == norm_url(self.cfg.public_url):
            return
        if u in ("http://127.0.0.1:%d" % self.cfg.port, "http://localhost:%d" % self.cfg.port):
            return
        self.peers[u] = Peer(u)
        if not quiet:
            self.log("added peer " + u)

    def hello_all(self):
        if not self.cfg.public_url:
            return
        for p in list(self.peers.values()):
            if p.kind == "node" and p.ok:
                try:
                    http_json(p.url + "/api/node/hello", {"url": norm_url(self.cfg.public_url)}, timeout=8)
                except Exception:
                    pass

    def sync_loop(self):
        last_hello = 0
        while not self.stop_ev.is_set():
            woke = self.wake.is_set(); self.wake.clear()
            t = time.time()
            for p in list(self.peers.values()):
                if self.stop_ev.is_set():
                    break
                interval = self.cfg.seed_interval if (p.seed or p.kind == "server") else self.cfg.peer_interval
                if not p.ok and p.err:
                    interval = max(interval, 15)
                if t - p.last_try >= interval or (woke and p.ok and p.kind == "server" and p.seed):
                    self.sync_peer(p)
            if woke:
                for p in list(self.peers.values()):
                    if p.kind == "node" and p.ok:
                        self.push(p)
            if time.time() - last_hello > 300:
                last_hello = time.time(); self.hello_all()
            if self.dirty and time.time() - self.last_save > 3:
                try:
                    self.save()
                except Exception as ex:
                    self.log("save failed: %s" % ex)
            self.status = self.mode()
            self.wake.wait(1.0)

    # ---------- API ----------
    def limit(self, ip, kind, mx):
        t = time.time()
        if len(self.rl) > 5000:
            self.rl.clear()
        k = kind + "|" + ip; e = self.rl.get(k)
        if not e or t - e[0] > 60:
            self.rl[k] = [t, 1]; return True
        e[1] += 1
        return e[1] <= mx

    def api_work(self, ch):
        f = self.following()
        if f:
            try:
                st, w = self.seed_work(f.url, ch, 1.5)
                if st == 429:
                    return 429, w
                if st == 200 and "y_candidate" in w:
                    with self.lock:
                        if not self.L.cand or self.L.cand["id"] != w["y_candidate"]:
                            self.wake.set()
                    w["via_node"] = NODE_VERSION
                    return 200, w
            except Exception:
                pass
        with self.lock:
            return 200, self.L.work_template(ch)

    def api_submit(self, ch, b):
        if not isinstance(b, dict) or b.get("chain") != ch:
            return 400, {"ok": False, "reason": "this endpoint accepts %s-Chain blocks only" % ch}
        f = self.following()
        if f:
            with self.lock:
                if self.L.cand and b.get("y_candidate") == self.L.cand["id"]:
                    why = self.L.check_evidence(b, "local", skip_time=True)  # the seed checks the time against its own clock
                else:
                    why = check_block(b, ch) if is_addr(b.get("miner")) else "miner must be a wallet address (eddy1 + 40 hex)"
            if why:
                self.stats["rejected"] += 1
                return 400, {"ok": False, "reason": why}
            try:
                st, r = http_json("%s/api/submit/%s" % (f.url, ch.lower()), b, timeout=20)
            except Exception as ex:
                return 503, {"ok": False, "reason": "seed unreachable (%s) - retry; the node continues independently after %d s" % (str(ex)[:60], self.cfg.seed_timeout)}
            self.work_cache.clear()
            if r.get("ok"):
                self.stats["accepted"] += 1; self.log("checked %s block #%s and passed it to the seed: accepted" % (ch, b.get("height")))
            else:
                self.stats["rejected"] += 1
            self.wake.set()
            return st, r
        with self.lock:
            L = self.L; L.open_candidate()
            try:
                rt = L.add_evidence(b, "local")
                cp = L.finalize_local() if (L.cand["e"] and L.cand["d"]) else None
                if cp:
                    L.open_candidate()
            except Invalid as ex:
                self.stats["rejected"] += 1
                return 400, {"ok": False, "reason": str(ex)}
            self.stats["accepted"] += 1
            self.log("accepted %s block #%d from a miner%s" % (ch, b["height"], (" -> Y checkpoint #%d finalized" % cp["height"]) if cp else ""))
            self.changed()
            return 200, {"ok": True, "height": b["height"], "retarget": rt,
                         "checkpoint": {"height": cp["height"], "id": cp["id"], "subsidy": cp["subsidy"], "e_to": cp["e_to"], "d_to": cp["d_to"]} if cp else None}

    def api_tx(self, body, relay=False):
        if not isinstance(body, dict) or not isinstance(body.get("tx"), dict):
            return 400, {"ok": False, "error": "expected { tx }"}
        if body.get("sell"):
            return 400, {"ok": False, "error": "selling to the sandbox exchange desk works only on the main server"}
        t = body["tx"]
        with self.lock:
            L = self.L; tid = t.get("txid")
            if tid in L.ytx or any(x["txid"] == tid for x in L.mempool) or (L.cand and tid in L.cand["txids"]):
                return 200, {"ok": True, "txid": tid, "status": "known"}
            why = validate_transfer(L.utxos, t)
        if why:
            return 400, {"ok": False, "error": "rejected: " + why}
        entry = {"txid": t["txid"], "from": address_of(t["pubkey"]), "amount": t["outputs"][0]["amount"], "time": now_ms(), "note": ""}
        f = self.following()
        if f and not relay:
            try:
                st, r = http_json(f.url + "/api/tx", {"tx": t}, timeout=20)
            except Exception as ex:
                return 503, {"ok": False, "error": "seed unreachable: %s" % str(ex)[:80]}
            if r.get("ok"):
                with self.lock:
                    self.L.registry.append(entry)
                self.wake.set()
            return st, r
        with self.lock:
            L = self.L
            why = validate_transfer(L.utxos, t)
            if why:
                return 400, {"ok": False, "error": "rejected: " + why}
            apply_tx(L.utxos, t); L.mempool.append(t)
            L.registry.append(entry); del L.registry[:-200]
            self.changed()
        self.log("accepted tx %s%s" % (t["txid"][:12], " (relayed by a node)" if relay else ""))
        if not f:
            threading.Thread(target=self.gossip_tx, args=(t,), daemon=True).start()
        return 200, {"ok": True, "txid": t["txid"], "status": "unconfirmed"}

    def gossip_tx(self, t):
        for p in list(self.peers.values()):
            if p.kind == "node" and p.ok:
                try:
                    http_json(p.url + "/api/tx", {"tx": t, "relay": True}, timeout=10)
                except Exception:
                    pass

    def api_state(self, addr):
        with self.lock:
            L = self.L
            slimB = lambda b: {k: b.get(k) for k in ("chain", "height", "hash", "prev", "y_candidate", "miner", "nonce", "bits", "time", "recv")}
            slimY = lambda cp: {"height": cp["height"], "id": cp["id"], "prev": cp["prev"], "y_candidate": cp["y_candidate"], "e": cp["e_evidence"], "d": cp["d_evidence"],
                                "state_root": cp["state_root"], "issued": cp["issued"], "subsidy": cp["subsidy"], "fees": cp["fees"], "time": cp["time"],
                                "txs": [{"txid": t["txid"], "type": t["type"], "amount": str(sum(amt_int(o["amount"]) for o in t["outputs"])), "fee": t["fee"]} for t in cp["txs"]],
                                "e_to": cp["e_to"], "d_to": cp["d_to"]}
            c = L.cand; h = c["height"] if c else len(L.Y)
            at, nr = next_reduction(h)
            cand = None
            if c:
                se, sd = reward_split(subsidy_at(c["height"], L.issued), sum(amt_int(t["fee"]) for t in c["txs"]))
                cand = {"height": c["height"], "id": c["id"], "txids": c["txids"], "e": c["e"], "d": c["d"], "e_miner": c["e_miner"], "d_miner": c["d_miner"],
                        "waiting": "D" if c["e"] and not c["d"] else "E" if c["d"] and not c["e"] else None, "reward": {"e": str(se), "d": str(sd)}}
            return {"network": NETWORK, "label": "Eddycoin Node (experimental)", "version": 2, "node": self.node_info(), "now": now_ms(), "paused": False,
                    "bits": L.bits, "live": {"target_ms": TARGET_MS, "retarget_every": RETARGET_EVERY, "clamp": CLAMP, "fork_height": FORK_H,
                                             "skew_ms": FORK_SKEW_MS if fork_active(c["height"] if L.cand else len(L.Y)) else MAX_SKEW_MS}, "retargets": L.retargets[-6:],
                    "tips": {"E": L.tip("E")["height"], "D": L.tip("D")["height"], "Y": len(L.Y) - 1},
                    "E": [slimB(b) for b in L.E[-14:]], "D": [slimB(b) for b in L.D[-14:]], "Y": [slimY(cp) for cp in L.Y[-12:]], "cand": cand,
                    "mempool": [t["txid"] for t in L.mempool],
                    "emission": {"R0": str(EM_R0), "epoch": EM_EPOCH, "num": str(EM_NUM), "den": str(EM_DEN), "height": h, "reward": str(emission_reward(h)), "next_at": at,
                                 "next_reward": str(nr), "coin": COIN, "decimals": DEC, "amounts": "decimal-string"},
                    "stats": {"issued": str(L.issued), "max_supply": str(MAX_SUPPLY), "subsidy": str(EM_R0), "checkpoints": len(L.Y) - 1, "blocks": {"E": len(L.E), "D": len(L.D)}, "anchor": -1},
                    "me": address_view(L, addr) if is_addr(addr or "") else None}

    def node_info(self):
        L = self.L
        return {"node": NODE_VERSION, "network": NETWORK, "genesis": GENESIS_ID, "tipY": len(L.Y) - 1, "tip_id": L.Y[-1]["id"],
                "tips": {"E": L.tip("E")["height"], "D": L.tip("D")["height"], "Y": len(L.Y) - 1}, "bits": L.bits, "work": str(L.work), "issued": str(L.issued),
                "cand_id": L.cand["id"] if L.cand else None, "cand_e": L.cand["e"] if L.cand else None, "cand_d": L.cand["d"] if L.cand else None,
                "mode": self.status, "public_url": norm_url(self.cfg.public_url) if self.cfg.public_url else None, "peers": self.public_peers()}

    def public_peers(self):
        t = time.time()
        out = [u for u, ts in list(self.announced.items()) if t - ts < 6 * 3600]
        if self.cfg.public_url and norm_url(self.cfg.public_url):
            out.insert(0, norm_url(self.cfg.public_url))
        return out[:30]

    def api_verify(self):
        if self.verify_cache and time.time() - self.verify_cache["t"] < 15:
            return self.verify_cache["r"]
        with self.lock:
            Y, E, D = list(self.L.Y), list(self.L.E), list(self.L.D)
            cand = copy.deepcopy(self.L.cand); mem = list(self.L.mempool); work = self.L.work; issued = self.L.issued
        checks = []
        try:
            t = time.time()
            R = ledger_from_parts(Y, E, D, cand, mem, self.cfg.genesis_bits)
            ok = R.work == work and R.issued == issued and R.Y[-1]["id"] == Y[-1]["id"]
            checks.append({"name": "full replay from genesis: PoW, retargeting, Y candidates, checkpoint ids, ECDSA signatures, UTXO / no double spends, emission",
                           "result": "PASS" if ok else "FAIL", "detail": "%d checkpoints, %d E + %d D blocks, %.1f s" % (len(Y) - 1, len(E), len(D), time.time() - t)})
            checks.append({"name": "issuance cap (900,000,000 EDDY)", "result": "PASS" if R.issued <= MAX_SUPPLY else "FAIL", "detail": "issued %s EDDY" % fmt(R.issued)})
        except Invalid as ex:
            checks.append({"name": "full replay from genesis", "result": "FAIL", "detail": str(ex)})
        r = {"overall": "PASS" if all(c["result"] == "PASS" for c in checks) else "FAIL", "checks": checks, "issued": str(issued), "time": now_ms(), "node": NODE_VERSION}
        self.verify_cache = {"t": time.time(), "r": r}
        return r

    def api_push(self, body, ip=None):
        if not isinstance(body, dict):
            return 400, {"ok": False, "error": "bad body"}
        if ip and self.is_banned("push@" + ip):
            return 403, {"ok": False, "error": "pushes from this address are refused for a while (invalid data)"}
        u = norm_url(body.get("from_url") or "") if isinstance(body.get("from_url"), str) else None
        ch, need, why = self.consider(body, prefer=False, source=u or "a node (push)")
        if why.startswith("INVALID"):
            self.log("REJECTED pushed data: %s" % why); self.on_invalid(u, body.get("from"), why, push_ip=ip or "?")
        elif ch:
            self.net_event(u, "data")
        return 200, {"ok": True, "changed": ch, "need_from": need, "result": why, "work": str(self.L.work)}

    def api_hello(self, body):
        u = norm_url(body.get("url")) if isinstance(body, dict) and isinstance(body.get("url"), str) else None
        if not u:
            return 400, {"ok": False, "error": "url must be http(s)://host[:port]"}
        def check():
            try:
                st, info = http_json(u + "/api/node/info", timeout=8)
                if st == 200 and info.get("genesis") == GENESIS_ID:
                    self.announced[u] = time.time(); self.add_peer(u)
            except Exception:
                pass
        threading.Thread(target=check, daemon=True).start()
        return 200, {"ok": True, "checking": u}

    # ---------- start / stop ----------
    def start(self):
        self.load()
        self.httpd = ThreadingHTTPServer((self.cfg.listen, self.cfg.port), make_handler(self))
        self.httpd.daemon_threads = True
        threading.Thread(target=self.httpd.serve_forever, daemon=True).start()
        threading.Thread(target=self.sync_loop, daemon=True).start()
        threading.Thread(target=self.xc_loop, daemon=True).start()
        self.log("node listening on http://%s:%d - miners / wallet: http://127.0.0.1:%d" % (self.cfg.listen, self.cfg.port, self.cfg.port))
        if self.cfg.listen in ("0.0.0.0", "::"):
            for ip in lan_ips():
                self.log("other computers in your network can use http://%s:%d" % (ip, self.cfg.port))

    def stop(self):
        self.stop_ev.set(); self.wake.set(); self.xc_wake.set()
        if self.httpd:
            self.httpd.shutdown(); self.httpd.server_close()
        try:
            self.save(force=True)
        except Exception as ex:
            self.log("save failed: %s" % ex)
        self.log("node stopped")

    def urls(self):
        out = ["http://127.0.0.1:%d" % self.cfg.port]
        if self.cfg.listen in ("0.0.0.0", "::"):
            out += ["http://%s:%d" % (ip, self.cfg.port) for ip in lan_ips()]
        if self.cfg.public_url and norm_url(self.cfg.public_url):
            out.append(norm_url(self.cfg.public_url) + " (public)")
        return out


STATUS_HTML = """<!doctype html><html><head><meta charset="utf-8"><title>Eddycoin Node</title><meta name="viewport" content="width=device-width,initial-scale=1">
<style>body{font:15px system-ui,sans-serif;background:#0b1020;color:#e6ebf5;max-width:760px;margin:30px auto;padding:0 14px}code,pre{background:#121a30;padding:2px 6px;border-radius:6px}a{color:#34d399}</style></head>
<body><h1>Eddycoin Node <small style="color:#9aa6c0">(experimental, test coin only)</small></h1><pre id="s">loading...</pre>
<p>Point the CPU / GPU miner or the wallet at <code>%(url)s</code> (field "Server"). API: <a href="/api/state">/api/state</a> &middot; <a href="/api/verify">/api/verify</a> &middot; <a href="/api/node/info">/api/node/info</a> &middot; <a href="/api/peers">/api/peers</a></p>
<script>async function r(){try{const i=await (await fetch('/api/node/info')).json();document.getElementById('s').textContent='mode:  '+i.mode+'\\nY-Chain checkpoints: #'+i.tipY+'\\nE-Chain tip: #'+i.tips.E+' ('+i.bits.E+' bits)\\nD-Chain tip: #'+i.tips.D+' ('+i.bits.D+' bits)\\nwork:  '+i.work}catch(e){}}r();setInterval(r,3000)</script></body></html>"""


def make_handler(node):
    class H(BaseHTTPRequestHandler):
        server_version = NODE_VERSION
        sys_version = ""
        protocol_version = "HTTP/1.0"

        def log_message(self, *a):
            pass

        def send(self, code, obj, ctype="application/json; charset=utf-8", extra=None):
            data = obj.encode("utf-8") if isinstance(obj, str) else json.dumps(obj, separators=(",", ":")).encode("utf-8")
            self.send_response(code); self.send_header("content-type", ctype); self.send_header("cache-control", "no-store")
            self.send_header("content-length", str(len(data)))
            for k, v in (extra or {}).items():
                self.send_header(k, v)
            self.end_headers(); self.wfile.write(data)

        def body(self, mx):
            n = int(self.headers.get("content-length") or 0)
            if n > mx:
                raise ValueError("body too large")
            raw = self.rfile.read(n) if n else b""
            return json.loads(raw.decode("utf-8")) if raw else None

        def do_GET(self):
            try:
                u = urllib.parse.urlsplit(self.path); p = u.path.rstrip("/"); q = urllib.parse.parse_qs(u.query); ip = self.client_address[0]
                g = lambda k: (q.get(k) or [""])[0]
                if p == "":
                    return self.send(200, STATUS_HTML % {"url": "http://127.0.0.1:%d" % node.cfg.port}, "text/html; charset=utf-8")
                if p == "/api/work":
                    if not node.limit(ip, "work", 240):
                        return self.send(429, {"error": "rate limit"})
                    ch = g("chain").upper()
                    if ch not in ("E", "D"):
                        return self.send(400, {"error": "use /api/work?chain=E or ?chain=D"})
                    return self.send(*node.api_work(ch))
                if p == "/api/state":
                    if not node.limit(ip, "state", 240):
                        return self.send(429, {"error": "rate limit"})
                    return self.send(200, node.api_state(g("addr")))
                if p == "/api/wallet" or p.startswith("/api/wallet/"):
                    if not node.limit(ip, "wallet", 240):
                        return self.send(429, {"error": "rate limit"})
                    a = g("addr") or p[len("/api/wallet/"):]
                    if not is_addr(a):
                        return self.send(400, {"error": "addr must be eddy1 + 40 hex"})
                    with node.lock:
                        v = address_view(node.L, a); v.update(decimals=DEC, network=NETWORK, cand_height=node.L.cand["height"] if node.L.cand else None)
                    return self.send(200, v)
                if p.startswith("/api/block/"):
                    parts = p.split("/")
                    if len(parts) != 5 or parts[3] not in ("E", "D", "Y") or not parts[4].isdigit():
                        return self.send(400, {"error": "use /api/block/E|D|Y/<height>"})
                    h = int(parts[4])
                    with node.lock:
                        arr = getattr(node.L, parts[3]); b = arr[h] if h < len(arr) else None
                    return self.send(200, b) if b else self.send(404, {"error": "not found"})
                if p == "/api/verify":
                    if not node.limit(ip, "verify", 6):
                        return self.send(429, {"error": "rate limit: 6 verifications per minute"})
                    return self.send(200, node.api_verify())
                if p == "/api/export":
                    if not node.limit(ip, "export", 10):
                        return self.send(429, {"error": "rate limit"})
                    with node.lock:
                        s = json.dumps(export_chain(node.L), separators=(",", ":"))
                    return self.send(200, s, extra={"content-disposition": "attachment; filename=eddycoin_chain_export.json"})
                if p == "/api/peers":
                    return self.send(200, {"peers": node.public_peers(), "note": "only nodes that chose to announce a public URL are listed"})
                if p == "/api/node/info":
                    if not node.limit(ip, "info", 240):
                        return self.send(429, {"error": "rate limit"})
                    with node.lock:
                        return self.send(200, node.node_info())
                if p == "/api/node/checkpoints":
                    if not node.limit(ip, "ckp", 120):
                        return self.send(429, {"error": "rate limit"})
                    try:
                        if g("h"):
                            hs = [int(x) for x in g("h").split(",") if x.strip().isdigit()][:40]
                        else:
                            f, t = int(g("from") or "0"), int(g("to") or "0")
                            hs = list(range(max(0, f), max(0, min(t, f + 199)) + 1))
                    except ValueError:
                        return self.send(400, {"error": "use ?from=<h>&to=<h> (max 200) or ?h=1,2,3 (max 40)"})
                    with node.lock:
                        r = {"network": NETWORK, "genesis": GENESIS_ID, "tipY": len(node.L.Y) - 1, "tip_id": node.L.Y[-1]["id"], "work": str(node.L.work),
                             "ids": node.ckp_ids(hs), "format": "height -> [checkpoint id, E evidence hash, D evidence hash]"}
                    return self.send(200, r)
                if p == "/api/node/mismatches":
                    if not node.limit(ip, "mm", 60):
                        return self.send(429, {"error": "rate limit"})
                    return self.send(200, node.api_mismatches())
                if p == "/api/node/segment":
                    if not node.limit(ip, "segment", 120):
                        return self.send(429, {"error": "rate limit"})
                    with node.lock:
                        s = json.dumps(node.segment(int(g("from") or "1")), separators=(",", ":"))
                    return self.send(200, s)
                return self.send(404, {"error": "not found", "endpoints": ["GET /api/work?chain=E|D", "POST /api/submit/e|d", "GET /api/wallet?addr=", "POST /api/tx",
                                                                     "GET /api/state", "GET /api/verify", "GET /api/export", "GET /api/peers", "GET /api/node/info",
                                                                     "GET /api/node/checkpoints?from=&to=", "GET /api/node/mismatches", "POST /api/node/report"]})
            except Exception as ex:
                try:
                    self.send(500, {"error": "node error: %s" % str(ex)[:200]})
                except Exception:
                    pass

        def do_POST(self):
            try:
                p = urllib.parse.urlsplit(self.path).path.rstrip("/"); ip = self.client_address[0]
                spec = {"/api/submit/e": ("submit", 60, 2048), "/api/submit/d": ("submit", 60, 2048), "/api/tx": ("tx", 30, 8192),
                        "/api/node/push": ("push", 120, 32 * 1024 * 1024), "/api/node/hello": ("hello", 5, 512), "/api/node/report": ("report", 10, 2048)}.get(p)
                if not spec:
                    return self.send(404, {"error": "not found"})
                if not node.limit(ip, spec[0], spec[1]):
                    return self.send(429, {"ok": False, "reason": "rate limit", "error": "rate limit"})
                try:
                    b = self.body(spec[2])
                except Exception as ex:
                    return self.send(400, {"ok": False, "reason": str(ex), "error": str(ex)})
                if p.startswith("/api/submit/"):
                    return self.send(*node.api_submit(p[-1].upper(), b))
                if p == "/api/tx":
                    return self.send(*node.api_tx(b, relay=bool(isinstance(b, dict) and b.get("relay"))))
                if p == "/api/node/push":
                    return self.send(*node.api_push(b, ip))
                if p == "/api/node/report":
                    return self.send(*node.api_report(b, ip))
                return self.send(*node.api_hello(b))
            except Exception as ex:
                try:
                    self.send(500, {"error": "node error: %s" % str(ex)[:200]})
                except Exception:
                    pass
    return H


# ======================================================================================================
# config, command line, window
# ======================================================================================================
def default_data_dir():
    if os.name == "nt":
        return os.path.join(os.environ.get("APPDATA") or os.path.expanduser("~"), "EddycoinNode")
    return os.path.join(os.path.expanduser("~"), ".eddycoin_node")


class Cfg:
    def __init__(self, **kw):
        self.port = DEFAULT_PORT; self.listen = "127.0.0.1"; self.public_url = ""; self.seeds = [DEFAULT_SEED]; self.peers = []
        self.data = default_data_dir(); self.genesis_bits = dict(GENESIS_BITS); self.seed_timeout = 60; self.seed_interval = 8; self.peer_interval = 5
        self.__dict__.update(kw)


def parse_bits(s):
    e, d = [int(x) for x in s.split(",")]
    return {"E": e, "D": d}


def build_cfg(a):
    cfg = Cfg(port=a.port, listen=a.listen, public_url=a.public_url or "", data=a.data or default_data_dir(), seed_timeout=a.seed_timeout,
              seed_interval=a.seed_interval, peer_interval=a.peer_interval)
    cfg.seeds = [] if a.no_seed else (a.seed or [DEFAULT_SEED])
    cfg.peers = a.peer or []
    if a.genesis_bits:
        cfg.genesis_bits = parse_bits(a.genesis_bits)
    return cfg


def headless(cfg):
    node = Node(cfg, log=lambda m: print(m, flush=True))
    node.start()
    stop = threading.Event()
    try:
        import signal
        signal.signal(signal.SIGTERM, lambda *a: stop.set())
    except Exception:
        pass
    try:
        last = 0
        while not stop.is_set():
            stop.wait(1)
            if time.time() - last > 30:
                last = time.time()
                with node.lock:
                    L = node.L
                    print(time.strftime("%H:%M:%S ") + "[%s] Y #%d | E #%d | D #%d | bits E %d / D %d | peers ok %d/%d | issued %s EDDY" % (
                        node.status, len(L.Y) - 1, L.tip("E")["height"], L.tip("D")["height"], L.bits["E"], L.bits["D"],
                        sum(1 for p in node.peers.values() if p.ok), len(node.peers), fmt(L.issued)), flush=True)
    except KeyboardInterrupt:
        pass
    finally:
        node.stop()


def gui(a):
    import tkinter as tk
    from tkinter import ttk, messagebox
    BG, PANEL, FG, MUTED, ACC = "#0b1020", "#121a30", "#e6ebf5", "#9aa6c0", "#10b981"
    data = a.data or default_data_dir(); os.makedirs(data, exist_ok=True)
    sp = os.path.join(data, "node_settings.json")
    try:
        with open(sp, encoding="utf-8") as f:
            S = json.load(f)
    except Exception:
        S = {}
    root = tk.Tk(); root.title("Eddycoin Node - EXPERIMENTAL (test coin, no real money)"); root.geometry("940x700"); root.minsize(780, 580); root.configure(bg=BG)
    import tkinter.font as tkf
    mono = "Consolas" if "Consolas" in tkf.families() else "DejaVu Sans Mono" if "DejaVu Sans Mono" in tkf.families() else "Courier"
    sty = ttk.Style()
    try:
        sty.theme_use("clam")
    except Exception:
        pass
    sty.configure(".", background=BG, foreground=FG, fieldbackground=PANEL); sty.configure("Muted.TLabel", foreground=MUTED)
    sty.configure("Big.TLabel", font=("Segoe UI", 13, "bold"), foreground=ACC)
    sty.configure("Treeview", background=PANEL, fieldbackground=PANEL, foreground=FG)
    sty.configure("Start.TButton", font=("Segoe UI", 12, "bold"), padding=8, background=ACC, foreground="#04120c")
    sty.configure("Stop.TButton", font=("Segoe UI", 12, "bold"), padding=8, background="#ef4444", foreground="white")
    sty.map("Start.TButton", background=[("disabled", "#2b3a4f")]); sty.map("Stop.TButton", background=[("disabled", "#2b3a4f")])
    tk.Label(root, text="Eddycoin Node - for play / experimental. Test coin only, no real money.", bg="#3b2a05", fg="#fde68a").pack(fill="x")
    fr = ttk.Frame(root, padding=10); fr.pack(fill="both", expand=True)
    top = ttk.Frame(fr); top.pack(fill="x")
    ttk.Label(top, text="Port").grid(row=0, column=0, sticky="w")
    port_v = tk.StringVar(value=str(S.get("port", a.port))); ttk.Entry(top, textvariable=port_v, width=7).grid(row=0, column=1, sticky="w", padx=4)
    lan_v = tk.BooleanVar(value=S.get("lan", a.listen == "0.0.0.0"))
    ttk.Checkbutton(top, text="Allow other computers to connect to my node", variable=lan_v).grid(row=0, column=2, sticky="w", padx=10)
    ttk.Label(top, text="Public URL (optional)").grid(row=1, column=0, sticky="w", pady=4)
    pub_v = tk.StringVar(value=S.get("public_url", a.public_url or "")); ttk.Entry(top, textvariable=pub_v, width=46).grid(row=1, column=1, columnspan=2, sticky="w", padx=4)
    ttk.Label(top, text="only if your router forwards the port to this PC, e.g. http://myname.duckdns.org:8770", style="Muted.TLabel").grid(row=2, column=1, columnspan=3, sticky="w")
    ttk.Label(top, text="Seed + other nodes\n(one URL per line)").grid(row=3, column=0, sticky="nw", pady=4)
    peers_t = tk.Text(top, height=4, width=62, bg=PANEL, fg=FG, insertbackground=FG); peers_t.grid(row=3, column=1, columnspan=3, sticky="we", padx=4, pady=4)
    default_lines = ([] if a.no_seed else (a.seed or [DEFAULT_SEED])) + (a.peer or [])
    peers_t.insert("1.0", "\n".join(S.get("peers_text") or default_lines))
    ttk.Label(top, text="The first line that is the live server (%s) is used as the seed. Lines starting with 'seed ' are seeds too." % DEFAULT_SEED, style="Muted.TLabel").grid(row=4, column=1, columnspan=3, sticky="w")
    bf = ttk.Frame(fr); bf.pack(fill="x", pady=6)
    start_b = ttk.Button(bf, text="Start node", style="Start.TButton"); start_b.pack(side="left")
    stop_b = ttk.Button(bf, text="Stop", style="Stop.TButton"); stop_b.pack(side="left", padx=8); stop_b.state(["disabled"])
    url_v = tk.StringVar(value="not running"); ttk.Label(bf, textvariable=url_v, style="Muted.TLabel").pack(side="left", padx=10)
    st = ttk.Frame(fr); st.pack(fill="x", pady=4)
    mode_v = tk.StringVar(value="stopped"); ttk.Label(st, textvariable=mode_v, style="Big.TLabel").pack(anchor="w")
    h_v = tk.StringVar(value=""); ttk.Label(st, textvariable=h_v, font=("Segoe UI", 11)).pack(anchor="w")
    s_v = tk.StringVar(value=""); ttk.Label(st, textvariable=s_v, style="Muted.TLabel").pack(anchor="w")
    cols = ("peer", "type", "status", "tip")
    tree = ttk.Treeview(fr, columns=cols, show="headings", height=5)
    for c, wd in zip(cols, (380, 80, 300, 100)):
        tree.heading(c, text=c.capitalize()); tree.column(c, width=wd, anchor="w")
    tree.pack(fill="x", pady=4)
    log = tk.Text(fr, height=12, bg=PANEL, fg=FG, font=(mono, 9)); log.pack(fill="both", expand=True)
    q = queue.Queue(); state = {"node": None}

    def start():
        try:
            port = int(port_v.get()); assert 1 <= port <= 65535
        except Exception:
            messagebox.showerror("Eddycoin Node", "Port must be a number 1-65535"); return
        seeds, peers = [], []
        for l in peers_t.get("1.0", "end").splitlines():
            l = l.strip()
            if not l or l.startswith("#"):
                continue
            is_seed = l.lower().startswith("seed ")
            u = norm_url(l[5:] if is_seed else l)
            if not u:
                messagebox.showerror("Eddycoin Node", "Not a valid URL: %s" % l); return
            (seeds if is_seed or u == norm_url(DEFAULT_SEED) else peers).append(u)
        pub = pub_v.get().strip()
        if pub and not norm_url(pub):
            messagebox.showerror("Eddycoin Node", "Public URL must look like http://name:port"); return
        S.update(port=port, lan=lan_v.get(), public_url=pub, peers_text=[l.strip() for l in peers_t.get("1.0", "end").splitlines() if l.strip()])
        try:
            with open(sp, "w", encoding="utf-8") as f:
                json.dump(S, f, indent=1)
        except Exception:
            pass
        cfg = Cfg(port=port, listen="0.0.0.0" if lan_v.get() else "127.0.0.1", public_url=pub, data=data, seeds=seeds, peers=peers, seed_timeout=a.seed_timeout)
        if a.genesis_bits:
            cfg.genesis_bits = parse_bits(a.genesis_bits)
        node = Node(cfg, log=lambda m: q.put(m))
        def run():
            try:
                node.start()
            except OSError as ex:
                q.put("cannot start: %s (is port %d already in use?)" % (ex, port)); q.put(None); return
            state["node"] = node
        start_b.state(["disabled"]); stop_b.state(["!disabled"]); mode_v.set("starting - validating the saved chain ...")
        threading.Thread(target=run, daemon=True).start()

    def stop():
        n = state["node"]; state["node"] = None
        stop_b.state(["disabled"])
        if n:
            threading.Thread(target=lambda: (n.stop(), q.put(None)), daemon=True).start()
        else:
            q.put(None)

    start_b.config(command=start); stop_b.config(command=stop)

    def tick():
        try:
            while True:
                m = q.get_nowait()
                if m is None:
                    start_b.state(["!disabled"]); stop_b.state(["disabled"]); url_v.set("not running"); mode_v.set("stopped"); continue
                log.insert("end", m + "\n"); log.see("end")
                if int(log.index("end-1c").split(".")[0]) > 800:
                    log.delete("1.0", "200.0")
        except queue.Empty:
            pass
        n = state["node"]
        if n and n.L:
            with n.lock:
                L = n.L
                mode_v.set("Mode: " + n.status)
                h_v.set("Y-Chain checkpoint #%d    E-Chain #%d (%d bits)    D-Chain #%d (%d bits)    issued %s EDDY" % (
                    len(L.Y) - 1, L.tip("E")["height"], L.bits["E"], L.tip("D")["height"], L.bits["D"], fmt(L.issued)))
                s_v.set("blocks from miners: %d accepted / %d rejected   |   checkpoints from other nodes: %d   |   invalid data rejected: %d   |   reorgs: %d   |   data: %s" % (
                    n.stats["accepted"], n.stats["rejected"], n.stats["from_peers"], n.stats["invalid_from_peers"], n.stats["reorgs"], n.data))
            url_v.set("Miners / wallet:  " + "    ".join(n.urls()))
            tree.delete(*tree.get_children())
            for p in list(n.peers.values()):
                tree.insert("", "end", values=(p.url + ("  (seed)" if p.seed else ""), p.kind or "?", "OK" if p.ok else (p.err or "connecting ...")[:70],
                                               ("Y #%s" % p.info.get("tipY")) if p.info else "-"))
        root.after(1000, tick)

    def on_close():
        n = state["node"]
        if n:
            try:
                n.stop()
            except Exception:
                pass
        root.destroy()
    root.protocol("WM_DELETE_WINDOW", on_close)
    root.after(500, tick)
    if a.autostart:
        root.after(300, start)
    root.mainloop()


def main():
    if sys.stdout is None or sys.stderr is None:  # windowed .exe has no console: write output to a log file
        try:
            d = default_data_dir(); os.makedirs(d, exist_ok=True)
            f = open(os.path.join(d, "node_console.log"), "a", encoding="utf-8", buffering=1)
        except Exception:
            f = open(os.devnull, "w")
        sys.stdout = sys.stdout or f; sys.stderr = sys.stderr or f
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--headless", action="store_true", help="no window (Linux / Raspberry Pi / macOS / servers)")
    ap.add_argument("--port", type=int, default=DEFAULT_PORT, help="HTTP port (default %d)" % DEFAULT_PORT)
    ap.add_argument("--listen", default="127.0.0.1", help="127.0.0.1 = only this computer (default); 0.0.0.0 = other computers may connect too")
    ap.add_argument("--public-url", help="your node's public URL if others can reach it (router port forwarding + dynamic DNS); announced to other nodes")
    ap.add_argument("--seed", action="append", help="seed server/node URL (default %s); repeatable" % DEFAULT_SEED)
    ap.add_argument("--no-seed", action="store_true", help="do not use the live server as seed")
    ap.add_argument("--peer", action="append", help="another node's URL, e.g. http://192.168.1.20:8770; repeatable")
    ap.add_argument("--data", help="data folder (default %s)" % default_data_dir())
    ap.add_argument("--seed-timeout", type=int, default=60, help="seconds without the seed before the node continues on its own (default 60)")
    ap.add_argument("--seed-interval", type=float, default=8, help=argparse.SUPPRESS)
    ap.add_argument("--peer-interval", type=float, default=5, help=argparse.SUPPRESS)
    ap.add_argument("--genesis-bits", help=argparse.SUPPRESS)  # private local test networks only, e.g. 18,20
    ap.add_argument("--autostart", action="store_true", help="window mode: start the node right away")
    ap.add_argument("--verify-file", help="only re-validate a chain export file (e.g. from /api/export) and exit")
    a = ap.parse_args()
    if a.verify_file:
        with open(a.verify_file, encoding="utf-8") as f:
            X = json.load(f)
        t = time.time()
        try:
            L = ledger_from_parts(X["Y"], X["E"], X["D"], X.get("cand"), X.get("mempool") or [], parse_bits(a.genesis_bits) if a.genesis_bits else None)
        except Invalid as ex:
            print("FAIL: %s" % ex); sys.exit(1)
        print("PASS: %d checkpoints, %d E + %d D blocks, issued %s EDDY, next difficulty E %d / D %d bits (%.2f s)" % (
            len(L.Y) - 1, len(L.E), len(L.D), fmt(L.issued), L.bits["E"], L.bits["D"], time.time() - t))
        return
    if a.headless:
        return headless(build_cfg(a))
    try:
        import tkinter
        del tkinter
    except Exception:
        print("tkinter not available - starting without a window (--headless)")
        return headless(build_cfg(a))
    gui(a)


if __name__ == "__main__":
    main()