#!/usr/bin/env python3 """ Wire all DHARMA soul Engrams as peers to each other. Run after all soul Engrams are started (launch_dharma.sh or scripts/launch_dharma.sh). How DHARMA peer wiring works in the native Engram binary --------------------------------------------------------- The native runtime (engram-runtime-native) tracks peers using graph primitives: - "dharma:self" — DharmaSelf node (this instance's identity) - "dharma:peer:@" — DharmaPeer node per known peer - "dharma-relation" edge from "dharma:self" to the peer node, weight > 0 To register soul B as a peer of soul A, POST to A's /api/edges: from_id = "dharma:self" to_id = "dharma:peer:@http://localhost:" relation = "dharma-relation" weight = 1.0 _auth = A's API key The runtime auto-creates DharmaSelf and DharmaPeer nodes from the edge. The dharma_peers() el builtin returns all peers with weight > 0. 19 souls × 18/2 = 171 bidirectional pairs = 342 total edge POSTs. Idempotent: re-running reinforces existing edges. Usage: python3 wire_peers.py # wire all running souls python3 wire_peers.py --dry-run # show what would be done python3 wire_peers.py --status # check connectivity python3 wire_peers.py --soul richard-feynman # wire one soul only """ import argparse import json import sys from pathlib import Path import requests FORGE_DIR = Path("/Users/will/Development/neuron-technologies/forge") REGISTRY_FILE = FORGE_DIR / "registry.json" def load_registry() -> list[dict]: reg = json.loads(REGISTRY_FILE.read_text()) return reg["imprints"] def soul_url(soul: dict) -> str: return soul.get("engram_url", f"http://localhost:{soul['engram_port']}") def soul_api_key(soul: dict) -> str: """Per-soul API key: ntn--2026 (matching the running DHARMA infrastructure).""" if soul.get("engram_api_key"): return soul["engram_api_key"] return f"ntn-{soul['slug']}-2026" def peer_node_id(soul: dict) -> str: """ Build the DHARMA peer node ID for a soul. Format: "dharma:peer:@" The runtime uses the part before @ as the peer identifier and the part after @ as the routing URL for outbound messages. """ return f"dharma:peer:{soul['slug']}@{soul_url(soul)}" def check_soul_alive(soul: dict) -> bool: try: resp = requests.get(f"{soul_url(soul)}/stats", timeout=3) return resp.status_code == 200 except requests.exceptions.RequestException: return False def register_peer_on_host(host: dict, peer: dict, dry_run: bool = False) -> bool: """ Register peer as a DHARMA peer of host by posting a dharma-relation edge. Posts to host's /api/edges with host's API key. """ edge_url = f"{soul_url(host)}/api/edges" peer_id = peer_node_id(peer) api_key = soul_api_key(host) body = { "from_id": "dharma:self", "to_id": peer_id, "relation": "dharma-relation", "weight": 1.0, "_auth": api_key, } if dry_run: print(f" [DRY-RUN] POST {edge_url} to_id={peer_id}") return True try: resp = requests.post(edge_url, json=body, timeout=10) if resp.status_code in (200, 201): return True print(f" WARN: POST {edge_url} → {resp.status_code}: {resp.text[:200]}") return False except requests.exceptions.RequestException as exc: print(f" ERROR: POST {edge_url} failed: {exc}") return False def wire_all(souls: list[dict], dry_run: bool = False) -> dict: """Wire every soul as a peer of every other soul.""" total_ok = 0 total_fail = 0 pair_count = 0 n = len(souls) expected = n * (n - 1) // 2 for i in range(n): for j in range(i + 1, n): a = souls[i] b = souls[j] pair_count += 1 ok1 = register_peer_on_host(a, b, dry_run=dry_run) ok2 = register_peer_on_host(b, a, dry_run=dry_run) if ok1 and ok2: total_ok += 1 if pair_count % 30 == 0 or pair_count == expected: print(f" [{pair_count}/{expected}] {a['slug']} <-> {b['slug']}") else: total_fail += 1 print( f" [{pair_count}/{expected}] {a['slug']} <-> {b['slug']} " f"FAIL (A→B={ok1} B→A={ok2})" ) return {"pairs": pair_count, "ok": total_ok, "failed": total_fail} def get_peer_count(soul: dict) -> int: """Count DharmaPeer nodes in a soul's graph.""" try: url = f"{soul_url(soul)}/api/nodes" headers = {"Authorization": f"Bearer {soul_api_key(soul)}"} resp = requests.get(url, headers=headers, timeout=10) resp.raise_for_status() nodes = resp.json() if isinstance(nodes, list): return sum(1 for n in nodes if n.get("node_type") == "DharmaPeer") return 0 except requests.exceptions.RequestException: return -1 def cmd_wire(souls: list[dict], dry_run: bool) -> None: alive = [s for s in souls if check_soul_alive(s)] dead = [s for s in souls if not check_soul_alive(s)] if dead: print(f"\nWARN: {len(dead)} soul(s) not reachable:") for s in dead: print(f" - {s['slug']} ({soul_url(s)})") print(f"\nStart them: bash scripts/launch_dharma.sh") if not alive: sys.exit(1) print(f"\nProceeding with {len(alive)} reachable souls...\n") else: print(f"All {len(alive)} souls reachable.\n") n = len(alive) expected = n * (n - 1) // 2 print(f"Wiring {n} souls ({expected} pairs, {expected * 2} edge registrations)...") if dry_run: print("[DRY-RUN — no requests will be sent]\n") stats = wire_all(alive, dry_run=dry_run) print(f"\nWiring complete:") print(f" Pairs attempted : {stats['pairs']}") print(f" Succeeded : {stats['ok']}") print(f" Failed : {stats['failed']}") if not dry_run: print(f"\nVerify: python3 scripts/wire_peers.py --status") def cmd_status(souls: list[dict]) -> None: print("DHARMA swarm status:\n") for soul in souls: if not check_soul_alive(soul): print(f" {soul['slug']:32s} OFFLINE") continue peer_count = get_peer_count(soul) label = f"peers={peer_count}" if peer_count >= 0 else "(read error)" print(f" {soul['slug']:32s} online {label}") def main(): parser = argparse.ArgumentParser( description="Wire DHARMA soul Engrams as peers via dharma-relation edges" ) parser.add_argument("--dry-run", action="store_true") parser.add_argument("--status", action="store_true") parser.add_argument("--soul", metavar="SLUG") args = parser.parse_args() souls = load_registry() # Backfill missing engram_url for soul in souls: if "engram_url" not in soul: soul["engram_url"] = f"http://localhost:{soul['engram_port']}" if args.soul: all_souls = souls target = next((s for s in souls if s["slug"] == args.soul), None) if not target: print(f"ERROR: soul {args.soul!r} not found in registry") sys.exit(1) if args.status: cmd_status([target]) else: others = [s for s in all_souls if s["slug"] != args.soul] alive_others = [s for s in others if check_soul_alive(s)] if not check_soul_alive(target): print(f"ERROR: {args.soul} is offline at {soul_url(target)}") sys.exit(1) print(f"Wiring {args.soul} to {len(alive_others)} souls...\n") ok = fail = 0 for peer in alive_others: o1 = register_peer_on_host(target, peer, dry_run=args.dry_run) o2 = register_peer_on_host(peer, target, dry_run=args.dry_run) if o1 and o2: ok += 1 print(f" wired: {args.soul} <-> {peer['slug']}") else: fail += 1 print(f" FAIL: {args.soul} <-> {peer['slug']}") print(f"\nDone: ok={ok} failed={fail}") else: if args.status: cmd_status(souls) else: cmd_wire(souls, dry_run=args.dry_run) if __name__ == "__main__": main()