893 lines
37 KiB
Python
893 lines
37 KiB
Python
#!/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()
|