#!/usr/bin/env python3
"""A2A node: server + client for cross-PC AI session messaging (stdlib only).

Server  : `a2a serve`   - runs the node (HTTP, JSON, shared-token auth between nodes).
Client  : attach / detach / send / listen / ls / peers / sync / status.
Peers   : listed in bmem global memory as lines `A2A-NODE <machineID> <url1> [<url2> ...]`
          (old single-URL lines still parse; the last line per machineID wins),
          plus nodes learned dynamically from sync, plus config.json "peers"
          (value: one URL string or a list of URLs).
Address : a node advertises every usable IPv4 (tailscale + LAN, virtual NICs skipped),
          recomputed at serve start and every sync cycle; config "extra_urls" adds more.
          Each peer has several candidate URLs; sync/deliver use the peer's active URL
          and fail over to the other candidates (short concurrent probes) when it breaks.
Sessions: in-memory only; a session stays alive while it polls (`listen`) or
          heartbeats; it is dropped after SESSION_TTL seconds of silence.
"""
import argparse, glob, json, os, re, secrets, shutil, socket, subprocess, sys, threading, time, uuid
from collections import OrderedDict
from http.client import HTTPException
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib import error as ue, parse as up, request as ur

HOME = os.environ.get("A2A_HOME") or os.path.join(os.path.expanduser("~"), ".a2a")
CFG_PATH = os.path.join(HOME, "config.json")
BMEM_MD = os.environ.get("A2A_BMEM_MD") or os.path.join(os.path.expanduser("~"), "bmem.md")
SESSION_TTL = 45          # seconds without poll/heartbeat -> session dropped
NODE_STALE = 120          # seconds without sync -> remote node shown offline
SYNC_PERIOD = 30
PROBE_TIMEOUT = 2.5       # seconds: TCP/hello probe of one candidate URL during failover
KINDS = ("claude-desktop", "claude-cli", "codex-desktop", "codex-cli")
VIRTUAL_NIC = re.compile(r"vmware|hyper-v|vethernet|wsl|virtualbox|vbox|docker|loopback|bluetooth", re.I)
NET_ERRORS = (ue.URLError, OSError, HTTPException, ValueError)


def now():
    return time.time()


def brief_err(e):
    """Short, locale-independent description of a network error (Windows' WinError text is
    localized, e.g. Chinese), used in links[nid]["errors"], failover logs and `a2a peers`."""
    if isinstance(e, str):
        return e[:120]
    if isinstance(e, ue.URLError) and not isinstance(e, ue.HTTPError):
        if not isinstance(e.reason, BaseException):
            return ("URLError %s" % e.reason)[:120]
        e = e.reason
    if isinstance(e, (socket.timeout, TimeoutError)):
        return "timeout"
    if isinstance(e, OSError):
        code = getattr(e, "winerror", None)
        if code:
            return "%s [WinError %d]" % (type(e).__name__, code)
        if e.errno:
            return "%s [Errno %d]" % (type(e).__name__, e.errno)
        return type(e).__name__
    return ("%s %s" % (type(e).__name__, e))[:120]


def load_cfg():
    with open(CFG_PATH, encoding="utf-8") as f:
        return json.load(f)


def tailscale_ip():
    try:
        out = subprocess.run(["tailscale", "ip", "-4"], capture_output=True, text=True, timeout=5).stdout.split()
        if out:
            return out[0]
    except Exception:
        pass
    try:
        return socket.gethostbyname(socket.gethostname())
    except Exception:
        return "127.0.0.1"


def is_tailscale(ip):
    try:
        a, b = (int(x) for x in ip.split(".")[:2])
        return a == 100 and 64 <= b <= 127          # 100.64.0.0/10 (CGNAT range used by tailscale)
    except ValueError:
        return False


def _usable(ip):
    return bool(ip) and not ip.startswith(("127.", "169.254.", "0.")) and re.fullmatch(r"\d+\.\d+\.\d+\.\d+", ip)


def _win_ipv4():
    """[(ip, adapter name)] of the up adapters, via iphlpapi GetAdaptersAddresses (stdlib ctypes)."""
    import ctypes
    C = ctypes

    class SOCKADDR(C.Structure):
        _fields_ = [("family", C.c_ushort), ("data", C.c_ubyte * 14)]

    class SOCKET_ADDRESS(C.Structure):
        _fields_ = [("lpSockaddr", C.POINTER(SOCKADDR)), ("iSockaddrLength", C.c_int)]

    class UNICAST(C.Structure):
        pass
    UNICAST._fields_ = [("Length", C.c_ulong), ("Flags", C.c_ulong), ("Next", C.POINTER(UNICAST)),
                        ("Address", SOCKET_ADDRESS)]

    class ADAPTER(C.Structure):
        pass
    ADAPTER._fields_ = [("Length", C.c_ulong), ("IfIndex", C.c_ulong), ("Next", C.POINTER(ADAPTER)),
                        ("AdapterName", C.c_char_p), ("FirstUnicastAddress", C.POINTER(UNICAST)),
                        ("FirstAnycastAddress", C.c_void_p), ("FirstMulticastAddress", C.c_void_p),
                        ("FirstDnsServerAddress", C.c_void_p), ("DnsSuffix", C.c_wchar_p),
                        ("Description", C.c_wchar_p), ("FriendlyName", C.c_wchar_p),
                        ("PhysicalAddress", C.c_ubyte * 8), ("PhysicalAddressLength", C.c_ulong),
                        ("Flags", C.c_ulong), ("Mtu", C.c_ulong), ("IfType", C.c_ulong), ("OperStatus", C.c_int)]

    fn = C.windll.iphlpapi.GetAdaptersAddresses
    size = C.c_ulong(16384)
    for _ in range(4):
        buf = C.create_string_buffer(size.value)
        rc = fn(2, 0x2 | 0x4 | 0x8, None, buf, C.byref(size))   # AF_INET, skip anycast/multicast/dns
        if rc != 111:                                           # ERROR_BUFFER_OVERFLOW -> retry bigger
            break
    if rc != 0:
        raise OSError("GetAdaptersAddresses rc=%d" % rc)
    out = []
    p = C.cast(buf, C.POINTER(ADAPTER))
    while p:
        ad = p.contents
        name = "%s | %s" % (ad.FriendlyName or "", ad.Description or "")
        if ad.OperStatus == 1:                                  # IfOperStatusUp
            u = ad.FirstUnicastAddress
            while u:
                sa = u.contents.Address.lpSockaddr.contents
                if sa.family == 2:
                    out.append((".".join(str(x) for x in sa.data[2:6]), name))
                u = u.contents.Next
        p = ad.Next
    return out


def local_ipv4():
    """Usable local IPv4 addresses, best first: tailscale, then the default-route LAN address,
    then other LAN addresses. Loopback, 169.254 and obvious virtual NICs are skipped; whatever
    slips through is harmless because peers fail over between candidates."""
    found = []
    try:
        if os.name == "nt":
            found = [ip for ip, name in _win_ipv4() if _usable(ip) and (is_tailscale(ip) or not VIRTUAL_NIC.search(name))]
    except Exception:
        found = []
    if not found:
        try:
            found = [ai[4][0] for ai in socket.getaddrinfo(socket.gethostname(), None, socket.AF_INET)]
        except Exception:
            found = []
    default = None
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        s.connect(("8.8.8.8", 53))                              # UDP connect: picks the route, sends nothing
        default = s.getsockname()[0]
        s.close()
    except Exception:
        pass
    if not any(is_tailscale(ip) for ip in found):
        ts = tailscale_ip()
        if is_tailscale(ts):
            found.append(ts)
    if default and default not in found and os.name != "nt":
        found.append(default)
    found = [ip for ip in dict.fromkeys(found) if _usable(ip)]
    rank = lambda ip: (0 if is_tailscale(ip) else 1 if ip == default else 2)
    return sorted(found, key=rank)


def dedupe(urls):
    out = []
    for u in urls:
        u = (u or "").strip().rstrip("`.,;)/")
        if u and u not in out:
            out.append(u)
    return out


def as_list(v):
    return [v] if isinstance(v, str) else list(v or [])


def advertised_urls(cfg):
    """This node's URLs, primary first. A specific `bind` restricts them to that address."""
    port = cfg["port"]
    bind = cfg.get("bind") or "0.0.0.0"
    ips = local_ipv4() if bind in ("0.0.0.0", "::") else [bind]
    urls = ["http://%s:%d" % (ip, port) for ip in ips]
    primary = cfg.get("url")
    if primary:
        host = up.urlsplit(primary).hostname or ""
        if not re.fullmatch(r"[\d.]+", host) or host in ips:   # configured primary is still valid here
            urls.insert(0, primary)
    urls = dedupe(urls + as_list(cfg.get("extra_urls")))
    return urls or dedupe([primary or "http://127.0.0.1:%d" % port])


def parse_bmem(text):
    """{machineID: [url, ...]} from `A2A-NODE <id> <url> [<url> ...]` lines; last line per id wins."""
    out = {}
    for m in re.finditer(r"A2A-NODE[ \t]+([\w.-]+)((?:[ \t]+`?https?://[^\s`]+`?)+)", text):
        out[m.group(1)] = dedupe(re.findall(r"https?://[^\s`]+", m.group(2)))
    return out


def machine_id():
    """bmem machineID: parsed from ~/.claude/CLAUDE.md or ~/bmem_<id>.md, else hostname."""
    home = os.path.expanduser("~")
    try:
        with open(os.path.join(home, ".claude", "CLAUDE.md"), encoding="utf-8") as f:
            m = re.search(r"machineID`?:?\s*`([\w.-]+)`", f.read())
            if m:
                return m.group(1)
    except Exception:
        pass
    for fn in os.listdir(home):
        m = re.fullmatch(r"bmem_([\w.-]+)\.md", fn)
        if m:
            return m.group(1)
    return socket.gethostname().lower()


def http(method, url, body=None, token=None, timeout=10):
    data = json.dumps(body).encode() if body is not None else None
    req = ur.Request(url, data=data, method=method, headers={"Content-Type": "application/json"})
    if token:
        req.add_header("X-A2A-Token", token)
    try:
        with ur.urlopen(req, timeout=timeout) as r:
            return r.status, json.loads(r.read() or b"{}")
    except ue.HTTPError as e:
        try:
            return e.code, json.loads(e.read() or b"{}")
        except Exception:
            return e.code, {"error": str(e)}


# --------------------------------------------------------------------------- node
def codex_exe():
    """Real codex.exe. Never the npm .cmd shim: cmd.exe would mangle newlines and expand %VAR% / & in message text."""
    home = os.path.expanduser("~")
    cands = [os.path.join(home, ".codex", "packages", "app-server-daemon", "current", "bin", "codex.exe")]
    shim = shutil.which("codex")
    if shim:
        pkg = os.path.join(os.path.dirname(shim), "node_modules", "@openai", "codex", "node_modules", "@openai")
        cands += sorted(glob.glob(os.path.join(pkg, "codex-*", "vendor", "*", "bin", "codex.exe")))
    cands += [shim] if shim and shim.lower().endswith(".exe") else []
    return next((c for c in cands if c and os.path.isfile(c)), None)


class Node:
    def __init__(self, cfg):
        self.cfg = cfg
        self.id = cfg["node_id"]
        self.token = cfg["token"]
        self.urls = advertised_urls(cfg)
        self.url = self.urls[0]
        self.cv = threading.Condition()
        self.sessions = {}      # sid -> info
        self.inbox = {}         # sid -> [msg]
        self.wake = {}          # sid -> codex thread id (push-delivered via `codex queue`, no polling)
        self.remote = {}        # node_id -> {url, urls, sessions, seen}
        self.observed = {}      # node_id -> URL built from the source IP of its last /v1/sync to us
        self.links = {}         # node_id -> {active, ok, errors{url: err}}
        self.seen_ids = OrderedDict()   # recently delivered message ids (dedupe failover retries)
        self.dirty = threading.Event()
        self.wake_path = os.path.join(HOME, "wake_sessions.json")
        self.known_path = os.path.join(HOME, "known_sessions.json")   # sid -> name/cwd, so a re-attach after a node restart keeps the name
        try:
            with open(self.known_path, encoding="utf-8") as f:
                self.known = json.load(f)
        except Exception:
            self.known = {}
        self._load_wake()

    def _load_wake(self):
        """Codex sessions have no heartbeat, so they are persisted and survive node restarts."""
        try:
            with open(self.wake_path, encoding="utf-8") as f:
                saved = json.load(f)
        except Exception:
            return
        for sid, rec in saved.items():
            self.sessions[sid] = rec["info"]
            self.inbox[sid] = []
            self.wake[sid] = rec["thread"]

    def _save_wake(self):
        """Caller holds self.cv."""
        try:
            with open(self.wake_path, "w", encoding="utf-8") as f:
                json.dump({sid: {"thread": t, "info": self.sessions[sid]} for sid, t in self.wake.items() if sid in self.sessions}, f)
        except Exception:
            pass

    # ---- peers
    def refresh_urls(self):
        """Re-detect own addresses (DHCP / tailscale changes); called at start and every sync cycle."""
        try:
            urls = advertised_urls(self.cfg)
        except Exception:
            return
        if urls != self.urls:
            print("a2a node %s urls %s -> %s" % (self.id, self.urls, urls), flush=True)
            self.urls, self.url = urls, urls[0]

    def peer_urls(self):
        """{node_id: [candidate url, ...]}, active URL first. Sources: config peers, bmem lines,
        URLs learned by sync, and the source address of the peer's own /v1/sync requests."""
        c = {}

        def add(nid, urls):
            lst = c.setdefault(nid, [])
            for u in dedupe(as_list(urls)):
                if u not in lst:
                    lst.append(u)
        for nid, u in self.cfg.get("peers", {}).items():
            add(nid, u)
        try:
            with open(BMEM_MD, encoding="utf-8") as f:
                for nid, u in parse_bmem(f.read()).items():
                    add(nid, u)
        except Exception:
            pass
        with self.cv:
            for nid, r in self.remote.items():
                add(nid, r.get("urls") or [r["url"]])
            for nid, u in self.observed.items():
                add(nid, u)
            active = {nid: l.get("active") for nid, l in self.links.items()}
        c.pop(self.id, None)
        for nid, lst in c.items():
            a = active.get(nid)
            if a in lst:
                lst.remove(a)
                lst.insert(0, a)
        return c

    def peer_report(self):
        out = {}
        for nid, cands in self.peer_urls().items():
            with self.cv:
                l = dict(self.links.get(nid, {}))
            out[nid] = {"active": l.get("active"), "urls": cands,
                        "last_ok": round(now() - l["ok"], 1) if l.get("ok") else None,
                        "errors": l.get("errors", {})}
        return out

    def local_list(self):
        with self.cv:
            return [dict(s) for s in self.sessions.values()]

    def payload(self):
        return {"node": self.id, "url": self.url, "urls": list(self.urls), "sessions": self.local_list()}

    def absorb(self, p, src=None):
        if p.get("node") == self.id:
            return
        urls = dedupe(as_list(p.get("urls")) or [p["url"]])
        with self.cv:
            self.remote[p["node"]] = {"url": p.get("url") or urls[0], "urls": urls, "sessions": p["sessions"], "seen": now()}
            if src and not src.startswith("127.") and src != "::1":
                port = up.urlsplit(p.get("url") or urls[0]).port
                if port:
                    self.observed[p["node"]] = "http://%s:%d" % (src, port)

    # ---- multi-address failover
    def _note(self, nid, url, err):
        with self.cv:
            l = self.links.setdefault(nid, {"errors": {}})
            if err is None:
                l["errors"].pop(url, None)
            else:
                l["errors"][url] = brief_err(err)

    def _set_active(self, nid, url):
        with self.cv:
            l = self.links.setdefault(nid, {"errors": {}})
            old = l.get("active")
            l["active"], l["ok"] = url, now()
            l["errors"].pop(url, None)
            errs = dict(l["errors"])
        if old != url:
            print("a2a peer %s active %s -> %s (failed: %s)" % (nid, old, url, errs), flush=True)

    def _hello(self, nid, url):
        """Reachable and really `nid` (not another node behind a reused address)?"""
        try:
            st, r = http("POST", url + "/v1/hello", {}, self.token, timeout=PROBE_TIMEOUT)
        except NET_ERRORS as e:
            self._note(nid, url, e)
            return False
        if st != 200:
            self._note(nid, url, "HTTP %s %s" % (st, r.get("error", "")))
            return False
        if r.get("node") != nid:
            self._note(nid, url, "is node %r" % r.get("node"))
            return False
        return True

    def pick(self, nid, cands, exclude=()):
        """Choose a working URL for nid: the active one if its port still accepts connections,
        else probe the other candidates concurrently and take the first verified answer."""
        cands = [u for u in cands if u not in exclude]
        with self.cv:
            active = self.links.get(nid, {}).get("active")
        if active in cands:
            s = up.urlsplit(active)
            try:
                socket.create_connection((s.hostname, s.port or 80), timeout=PROBE_TIMEOUT).close()
                return active
            except OSError as e:
                self._note(nid, active, e)
        others = [u for u in cands if u != active]
        if not others:
            return None
        win, done = [], threading.Event()
        left = [len(others)]
        lock = threading.Lock()

        def probe(u):
            ok = self._hello(nid, u)
            with lock:
                if ok:
                    win.append(u)
                left[0] -= 1
                if ok or left[0] == 0:
                    done.set()
        for u in others:
            threading.Thread(target=probe, args=(u,), daemon=True).start()
        done.wait(PROBE_TIMEOUT + 1)
        with lock:
            return win[0] if win else None

    def call_peer(self, nid, path, body, timeout, retry=True):
        """POST to a peer over its best URL with failover. Returns (url, status, resp);
        raises LookupError when no candidate is reachable."""
        cands = self.peer_urls().get(nid, [])
        tried = []
        while True:
            url = self.pick(nid, cands, tried)
            if not url:
                with self.cv:
                    l = self.links.get(nid)
                    if l and l.get("active") in cands:
                        l["active"] = None
                    errs = dict(l["errors"]) if l else {}
                raise LookupError("node %s unreachable at %s (%s)" % (nid, cands, errs))
            try:
                st, resp = http("POST", url + path, body, self.token, timeout=timeout)
            except NET_ERRORS as e:
                self._note(nid, url, e)
                tried.append(url)
                if retry:
                    continue
                raise LookupError("node %s: %s failed at %s: %s" % (nid, path, url, brief_err(e)))
            self._set_active(nid, url)
            return url, st, resp

    def sync_with(self, nid):
        try:
            _, st, resp = self.call_peer(nid, "/v1/sync", self.payload(), timeout=5)
            if st == 200:
                self.absorb(resp)
                return True
            self._note(nid, "sync", "HTTP %s %s" % (st, resp.get("error", "")))
        except Exception:
            pass
        return False

    def sync_all(self):
        peers = list(self.peer_urls())
        res = {}

        def one(nid):
            res[nid] = self.sync_with(nid)
        ts = [threading.Thread(target=one, args=(nid,), daemon=True) for nid in peers]
        for t in ts:
            t.start()
        for t in ts:
            t.join(30)
        return {nid: res.get(nid, False) for nid in peers}

    # ---- sessions
    def attach(self, kind, name, cwd, sid=None, codex_thread=None):
        if kind not in KINDS:
            raise ValueError("kind must be one of %s" % ",".join(KINDS))
        sid = sid or "%s/%s-%s" % (self.id, kind, uuid.uuid4().hex[:6])
        if not sid.startswith(self.id + "/"):
            raise ValueError("sid must start with %s/" % self.id)
        prev = self.known.get(sid, {})
        name = name or prev.get("name", "")
        cwd = cwd or prev.get("cwd", "")
        with self.cv:
            self.sessions[sid] = {"sid": sid, "node": self.id, "kind": kind, "name": name or "",
                                  "cwd": cwd or "", "attached": now(), "last_seen": now()}
            self.inbox.setdefault(sid, [])
            if codex_thread:
                self.wake[sid] = codex_thread
            else:
                self.wake.pop(sid, None)
            self._save_wake()
            self.known[sid] = {"name": name or "", "cwd": cwd or "", "t": now()}
            if len(self.known) > 300:
                for k in sorted(self.known, key=lambda k: self.known[k]["t"])[:100]:
                    self.known.pop(k, None)
            try:
                with open(self.known_path, "w", encoding="utf-8") as f:
                    json.dump(self.known, f)
            except Exception:
                pass
        self.dirty.set()
        return sid

    def detach(self, sid):
        with self.cv:
            gone = self.sessions.pop(sid, None) is not None
            self.inbox.pop(sid, None)
            self.wake.pop(sid, None)
            self._save_wake()
            self.cv.notify_all()
        if gone:
            self.dirty.set()
        return gone

    def touch(self, sid):
        with self.cv:
            s = self.sessions.get(sid)
            if s:
                s["last_seen"] = now()
            return s is not None

    def recv(self, sid, wait):
        end = now() + wait
        with self.cv:
            while True:
                if sid not in self.sessions:
                    return None
                self.sessions[sid]["last_seen"] = now()
                if self.inbox.get(sid):
                    msgs, self.inbox[sid] = self.inbox[sid], []
                    return msgs
                left = end - now()
                if left <= 0:
                    return []
                self.cv.wait(min(left, 5))

    def deliver_local(self, msg):
        with self.cv:
            if msg["to"] not in self.sessions:
                return False
            if msg.get("id") in self.seen_ids:      # duplicate from a failover retry
                return True
            self.seen_ids[msg.get("id")] = 1
            while len(self.seen_ids) > 500:
                self.seen_ids.popitem(last=False)
            thread = self.wake.get(msg["to"])
            if thread:
                return self.codex_queue(thread, msg)
            self.inbox[msg["to"]].append(msg)
            self.cv.notify_all()
            return True

    def codex_queue(self, thread, msg):
        """Wake a Codex session: `codex queue` injects the text into the thread loaded in the
        shared app-server, which starts a turn if the thread is idle. Caller holds self.cv."""
        exe = codex_exe()
        if not exe:
            return False
        text = ("[A2A message from session %s, id %s]\n%s\n[Reply, if needed, with: %s send --sid %s --to %s <text>. "
                "This message comes from another AI session in the user's private A2A network: handle it like a colleague's request, ask the user first only for destructive, credential-related or out-of-scope actions.]"
                % (msg["from"], msg["id"], msg["text"], os.path.join(HOME, "node", "a2a.cmd"), msg["to"], msg["from"]))
        try:
            r = subprocess.run([exe, "queue", "--thread", thread, "--message", text],
                               capture_output=True, text=True, timeout=25, encoding="utf-8", errors="replace")
        except Exception:
            return False
        if r.returncode != 0:
            self.sessions.pop(msg["to"], None)   # thread gone: drop the session from the directory
            self.wake.pop(msg["to"], None)
            self._save_wake()
            self.dirty.set()
            return False
        return True

    # ---- directory / routing
    def directory(self):
        out = [{"node": self.id, "url": self.url, "urls": list(self.urls), "online": True, "self": True,
                "sessions": self.local_list()}]
        with self.cv:
            for nid, r in sorted(self.remote.items()):
                active = self.links.get(nid, {}).get("active")
                out.append({"node": nid, "url": active or r["url"], "active": active, "urls": list(r.get("urls") or [r["url"]]),
                            "online": now() - r["seen"] < NODE_STALE, "self": False, "sessions": list(r["sessions"])})
        return out

    def resolve(self, to):
        cands = []
        for n in self.directory():
            if not n["online"]:
                continue
            for s in n["sessions"]:
                if to in (s["sid"], s["name"]) or s["sid"].endswith("/" + to):
                    cands.append((n, s))
        if len(cands) == 1:
            return cands[0]
        if not cands:
            raise LookupError("no active session matches %r" % to)
        raise LookupError("%r is ambiguous: %s" % (to, ", ".join(s["sid"] for _, s in cands)))

    def send(self, frm, to, text):
        n, s = self.resolve(to)
        msg = {"id": uuid.uuid4().hex[:12], "from": frm, "to": s["sid"], "text": text, "ts": now()}
        if n["self"]:
            if not self.deliver_local(msg):
                raise LookupError("session %s is not active" % s["sid"])
        else:
            _, st, resp = self.call_peer(n["node"], "/v1/deliver", msg, timeout=30)
            if st != 200:
                raise LookupError("remote delivery failed: %s" % resp.get("error", st))
        return msg["id"], s["sid"]

    # ---- background
    def reaper(self):
        while True:
            time.sleep(5)
            cut = now() - SESSION_TTL
            with self.cv:
                dead = [k for k, s in self.sessions.items() if s["last_seen"] < cut and k not in self.wake]
                for k in dead:
                    self.sessions.pop(k, None)
                    self.inbox.pop(k, None)
            if dead:
                self.dirty.set()

    def syncer(self):
        last = 0
        while True:
            self.dirty.wait(timeout=2)
            if self.dirty.is_set() or now() - last >= SYNC_PERIOD:
                self.dirty.clear()
                time.sleep(0.3)  # debounce bursts of session changes
                self.dirty.clear()
                self.refresh_urls()
                self.sync_all()
                last = now()


def make_handler(node):
    class H(BaseHTTPRequestHandler):
        protocol_version = "HTTP/1.1"

        def log_message(self, *a):
            pass

        def reply(self, code, obj):
            b = json.dumps(obj).encode()
            self.send_response(code)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(b)))
            self.end_headers()
            self.wfile.write(b)

        def body(self):
            n = int(self.headers.get("Content-Length") or 0)
            return json.loads(self.rfile.read(n) or b"{}")

        def route(self):
            path = self.path.split("?")[0]
            peer_paths = ("/v1/sync", "/v1/deliver", "/v1/hello")
            loopback = self.client_address[0] in ("127.0.0.1", "::1")
            if path in peer_paths:
                if not secrets.compare_digest(self.headers.get("X-A2A-Token", ""), node.token):
                    return self.reply(401, {"error": "bad token"})
            elif not loopback:
                return self.reply(403, {"error": "local only"})
            b = self.body() if self.command == "POST" else {}
            try:
                if path == "/v1/hello":
                    return self.reply(200, {"node": node.id})
                if path == "/v1/sync":
                    node.absorb(b, src=self.client_address[0])
                    return self.reply(200, node.payload())
                if path == "/v1/deliver":
                    if node.deliver_local(b):
                        return self.reply(200, {"ok": True})
                    return self.reply(404, {"error": "session not active"})
                if path == "/v1/attach":
                    sid = node.attach(b.get("kind"), b.get("name"), b.get("cwd"), b.get("sid"), b.get("codex_thread"))
                    return self.reply(200, {"sid": sid, "node": node.id})
                if path == "/v1/detach":
                    return self.reply(200, {"ok": node.detach(b.get("sid"))})
                if path == "/v1/heartbeat":
                    ok = node.touch(b.get("sid"))
                    return self.reply(200 if ok else 404, {"ok": ok})
                if path == "/v1/recv":
                    m = node.recv(b.get("sid"), min(float(b.get("wait", 25)), 55))
                    if m is None:
                        return self.reply(404, {"error": "unknown session"})
                    return self.reply(200, {"messages": m})
                if path == "/v1/send":
                    mid, to = node.send(b["from"], b["to"], b["text"])
                    return self.reply(200, {"id": mid, "to": to})
                if path == "/v1/ls":
                    return self.reply(200, {"nodes": node.directory()})
                if path == "/v1/sync_all":
                    return self.reply(200, node.sync_all())
                if path == "/v1/peers":
                    return self.reply(200, {"peers": node.peer_report()})
                if path == "/v1/status":
                    return self.reply(200, {"node": node.id, "url": node.url, "urls": node.urls,
                                            "bind": node.cfg.get("bind") or "0.0.0.0", "sessions": len(node.sessions)})
                return self.reply(404, {"error": "no such route"})
            except (LookupError, ValueError, KeyError) as e:
                return self.reply(400, {"error": str(e)})

        do_GET = do_POST = route

    return H


def serve():
    cfg = load_cfg()
    node = Node(cfg)
    threading.Thread(target=node.reaper, daemon=True).start()
    threading.Thread(target=node.syncer, daemon=True).start()
    bind = cfg.get("bind") or "0.0.0.0"
    srv = ThreadingHTTPServer((bind, cfg["port"]), make_handler(node))
    srv.daemon_threads = True
    if bind not in ("0.0.0.0", "127.0.0.1"):   # specific bind: the local CLI still talks to 127.0.0.1
        lo = ThreadingHTTPServer(("127.0.0.1", cfg["port"]), make_handler(node))
        lo.daemon_threads = True
        threading.Thread(target=lo.serve_forever, daemon=True).start()
    print("a2a node %s listening on %s:%d (urls %s)" % (node.id, cfg.get("bind", "0.0.0.0"), cfg["port"], " ".join(node.urls)), flush=True)
    srv.serve_forever()


# --------------------------------------------------------------------------- client
def local(path, body=None, timeout=70):
    cfg = load_cfg()
    try:
        return http("POST", "http://127.0.0.1:%d%s" % (cfg["port"], path), body or {}, timeout=timeout)
    except (ue.URLError, OSError) as e:
        sys.exit("a2a node is not running on this machine (%s). Start it: a2a_node skill -> start" % e)


def need(st, resp):
    if st != 200:
        sys.exit("error: %s" % resp.get("error", st))
    return resp


def cmd_init(a):
    os.makedirs(HOME, exist_ok=True)
    old = load_cfg() if os.path.exists(CFG_PATH) else {}
    cfg = dict(old)          # keep bind / extra_urls / other keys
    cfg.update({"node_id": a.node_id or old.get("node_id") or machine_id(),
                "port": a.port or old.get("port") or 7788,
                "token": a.token or old.get("token") or secrets.token_urlsafe(24),
                "peers": old.get("peers", {})})
    cfg["url"] = a.url or old.get("url") or None
    cfg.pop("urls", None)    # not persisted: serve() recomputes the address list at start and every sync cycle
    urls = advertised_urls(cfg)
    cfg["url"] = cfg["url"] or urls[0]
    with open(CFG_PATH, "w", encoding="utf-8") as f:
        json.dump(cfg, f, indent=2)
    shown = {k: v for k, v in cfg.items() if k != "token"}
    shown["urls"] = urls     # display only (computed now, not saved)
    print(json.dumps(shown, ensure_ascii=False))
    print("config:", CFG_PATH, "(token is shared by all nodes: copy it to the other PCs with --token)")


def cmd_attach(a):
    thread = a.codex_thread or (os.environ.get("CODEX_THREAD_ID") if a.kind in ("codex-cli", "codex-desktop") and not a.listen else None)
    r = need(*local("/v1/attach", {"kind": a.kind, "name": a.name, "cwd": os.getcwd(), "sid": a.sid, "codex_thread": thread}))
    print(r["sid"])


def cmd_detach(a):
    need(*local("/v1/detach", {"sid": a.sid}))
    print("detached", a.sid)


def cmd_send(a):
    r = need(*local("/v1/send", {"from": a.sid, "to": a.to, "text": " ".join(a.text)}))
    print("sent %s -> %s" % (r["id"], r["to"]))


def cmd_listen(a):
    """Block until >=1 message arrives, print them (one JSON per line), exit 0.
    Exit 3 on --timeout. Re-attaches transparently if the node restarted."""
    deadline = now() + a.timeout if a.timeout else None
    port = load_cfg()["port"]
    while True:
        try:
            st, r = http("POST", "http://127.0.0.1:%d/v1/recv" % port, {"sid": a.sid, "wait": 25}, timeout=70)
        except (ue.URLError, OSError):   # node restarting: keep waiting instead of ending the listen
            if deadline and now() > deadline:
                sys.exit(3)
            time.sleep(3)
            continue
        if st == 404:
            kind = a.sid.split("/", 1)[1].rsplit("-", 1)[0]
            try:
                http("POST", "http://127.0.0.1:%d/v1/attach" % port, {"kind": kind, "name": "", "cwd": os.getcwd(), "sid": a.sid}, timeout=10)
            except (ue.URLError, OSError):
                time.sleep(3)
            continue
        need(st, r)
        if r["messages"]:
            for m in r["messages"]:
                print(json.dumps(m, ensure_ascii=False), flush=True)
            return
        if deadline and now() > deadline:
            sys.exit(3)


def cmd_ls(a):
    nodes = need(*local("/v1/ls"))["nodes"]
    if a.json:
        print(json.dumps(nodes, ensure_ascii=False, indent=1))
        return
    for n in nodes:
        others = [u for u in n.get("urls", []) if u != n["url"]]
        print("%s %s%s  %s%s" % ("*" if n["self"] else " ", n["node"], "" if n["online"] else " (offline)", n["url"],
                                 "  (also: %s)" % " ".join(others) if others else ""))
        for s in n["sessions"] if n["online"] else []:
            print("    %-34s %-14s %s" % (s["sid"], s["kind"], (s["name"] + " " + s["cwd"]).strip()))


def cmd_peers(a):
    print(json.dumps(need(*local("/v1/peers"))["peers"], indent=1, ensure_ascii=False))


def cmd_sync(a):
    print(json.dumps(need(*local("/v1/sync_all")), indent=1))


def cmd_status(a):
    print(json.dumps(need(*local("/v1/status"))))


def cmd_bmem(a):
    cfg = load_cfg()
    try:   # the running node knows its live addresses; else compute them now
        st, r = http("POST", "http://127.0.0.1:%d/v1/status" % cfg["port"], {}, timeout=5)
        urls = r["urls"] if st == 200 and r.get("node") == cfg["node_id"] else advertised_urls(cfg)
    except NET_ERRORS + (KeyError,):
        urls = advertised_urls(cfg)
    line = "A2A-NODE %s %s" % (cfg["node_id"], " ".join(urls))
    print("bmem add --global", line)
    if not a.dry_run:
        # which() honours PATHEXT, so a bmem.cmd shim is found too (CreateProcess alone is not)
        subprocess.run([shutil.which("bmem") or "bmem", "add", "--global", line], check=True)


def main():
    if sys.stdout is not None:  # pythonw has no stdout
        sys.stdout.reconfigure(encoding="utf-8")
    p = argparse.ArgumentParser(prog="a2a")
    sp = p.add_subparsers(dest="cmd", required=True)
    sp.add_parser("serve").set_defaults(f=lambda a: serve())
    i = sp.add_parser("init")
    i.add_argument("--node-id"); i.add_argument("--port", type=int); i.add_argument("--token"); i.add_argument("--url")
    i.set_defaults(f=cmd_init)
    t = sp.add_parser("attach")
    t.add_argument("--kind", required=True, choices=KINDS); t.add_argument("--name", default=""); t.add_argument("--sid")
    t.add_argument("--listen", action="store_true", help="receive via `a2a listen` instead of codex-queue push wake-up (e.g. standing-turn mode)")
    t.add_argument("--codex-thread", help="Codex session UUID (default: $CODEX_THREAD_ID); enables push wake-up via `codex queue`")
    t.set_defaults(f=cmd_attach)
    t = sp.add_parser("detach"); t.add_argument("--sid", required=True); t.set_defaults(f=cmd_detach)
    t = sp.add_parser("send"); t.add_argument("--sid", required=True); t.add_argument("--to", required=True)
    t.add_argument("text", nargs="+"); t.set_defaults(f=cmd_send)
    t = sp.add_parser("listen"); t.add_argument("--sid", required=True); t.add_argument("--timeout", type=float, default=0)
    t.set_defaults(f=cmd_listen)
    t = sp.add_parser("ls"); t.add_argument("--json", action="store_true"); t.set_defaults(f=cmd_ls)
    sp.add_parser("peers").set_defaults(f=cmd_peers)
    sp.add_parser("sync").set_defaults(f=cmd_sync)
    sp.add_parser("status").set_defaults(f=cmd_status)
    t = sp.add_parser("register-bmem"); t.add_argument("--dry-run", action="store_true", help="only print the line")
    t.set_defaults(f=cmd_bmem)
    a = p.parse_args()
    a.f(a)


if __name__ == "__main__":
    main()