From 853313f95d4fd256ce0b2996036c1ad384b805a8 Mon Sep 17 00:00:00 2001 From: davidfarah2003 Date: Thu, 25 Jun 2026 23:17:01 -0700 Subject: [PATCH] feat(examples): add 05-scale-showcase (live agent-mesh galaxy) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A standalone, private example that spins up thousands of real Cotal endpoints over a bare NATS broker (live-only, no persistence) and renders them as a GPU galaxy. - swarm.ts: N real CotalEndpoints sharded across cluster workers; each agent DMs a stable peer set and posts to its team channel, with sparse cross-team bridges plus a few random long-range arcs (a small-world mesh). Gentle connect ramp + retry to survive large connect bursts. - observe.ts: a scale-safe read-only observer that taps the space and folds traffic into a graph model, emitting a bounded ~15Hz aggregated SSE feed (bytes scale with graph size, not message rate). Reports per-mode rates, avg delivery latency, and broker cpu/mem/conns via /varz. - web/: a cosmos.gl GPU galaxy (CDN-loaded) — role-coloured nodes, channel hubs, warm cross-team bridges, traffic heat pulses, hover and click-to-isolate, and a live HUD. - showcase.ts: one-command orchestrator (broker + observer + swarm). Depends only on @cotal-ai/core; no protocol or core changes. --- examples/05-scale-showcase/README.md | 58 ++++++ examples/05-scale-showcase/package.json | 20 ++ examples/05-scale-showcase/src/observe.ts | 226 +++++++++++++++++++++ examples/05-scale-showcase/src/showcase.ts | 64 ++++++ examples/05-scale-showcase/src/swarm.ts | 143 +++++++++++++ examples/05-scale-showcase/tsconfig.json | 8 + examples/05-scale-showcase/web/galaxy.html | 59 ++++++ examples/05-scale-showcase/web/galaxy.js | 212 +++++++++++++++++++ pnpm-lock.yaml | 10 + 9 files changed, 800 insertions(+) create mode 100644 examples/05-scale-showcase/README.md create mode 100644 examples/05-scale-showcase/package.json create mode 100644 examples/05-scale-showcase/src/observe.ts create mode 100644 examples/05-scale-showcase/src/showcase.ts create mode 100644 examples/05-scale-showcase/src/swarm.ts create mode 100644 examples/05-scale-showcase/tsconfig.json create mode 100644 examples/05-scale-showcase/web/galaxy.html create mode 100644 examples/05-scale-showcase/web/galaxy.js diff --git a/examples/05-scale-showcase/README.md b/examples/05-scale-showcase/README.md new file mode 100644 index 00000000..989992a9 --- /dev/null +++ b/examples/05-scale-showcase/README.md @@ -0,0 +1,58 @@ +# 05 · Scale showcase — the galaxy + +Thousands of **real** Cotal endpoints, live over NATS, rendered as a glowing GPU galaxy. + +This is a showcase, not a benchmark harness: it spins up *N* genuine `CotalEndpoint`s that +actually publish multicast / DM / anycast traffic, taps the mesh with one read-only observer, and +draws the whole thing as a force-directed galaxy: agents wired to the peers they DM and clustered +around the channel hubs they post on — team lobes joined by cross-team bridges, glowing as traffic +flows, with a live messages-delivered counter. + +```bash +pnpm demo --n 5000 # broker + observer + 5000 endpoints, then open the printed URL +pnpm demo --n 1000 # lighter +pnpm demo --n 12000 --workers 8 +``` + +Open **http://127.0.0.1:7900/**. Ctrl-C tears everything down. + +Measured on an 18-core / 48 GB Mac: **5,000 endpoints at 120 fps**, browser ≈ 0.6 GB, and the whole +mesh (broker + 4 swarm workers + observer) ≈ 1.2 GB. The protocol is cheap; the only thing that has +to be careful is the renderer. + +## Why it's built this way + +The naive version of this melts. Pointing the stock `cotal web` graph at just **500** live nodes +drove its Chrome renderer to **50 GB** and ~0.1 fps — because it spawns an animation object per +message (with channel fan-out) and the per-frame draw cost grows with traffic, so under load it +produces draw work faster than it can retire it. Unbounded by construction. + +This showcase fixes that with two rules: + +1. **Bounded feed (`src/observe.ts`).** The observer folds every message into an in-memory graph + model and emits a *fixed-rate* (~15 Hz) batched delta + a small "what lit up this frame" set. + The bytes sent to the browser scale with the **size of the graph**, never the message rate. The + structure is two overlaid graphs: the **DM peer mesh** (each agent DMs a stable set of + collaborators → agent↔agent edges, the cross-team bridges) and **channel hubs** (each agent + posts to its team channel → a hub node every team lobe gathers around). Anycast stays + *transient*: it pulses its sender but adds no edge, so the layout settles and stays stable. A + running counter tracks total messages seen on the wire. +2. **Bounded renderer (`web/galaxy.js`).** [cosmos.gl](https://github.com/cosmosgl/graph) runs the + force layout and rendering on the **GPU** with fixed, pre-allocated buffers. Pulses aren't + objects — an `act` frame just sets `heat = 1` on a few indices; the render loop decays heat and + uploads one node-color + one link-color buffer per frame. Per-frame work scales with what's + on screen, not with traffic. + +Net: work proportional to *what's visible*, not to *how much is happening*. + +## Pieces + +| File | Role | +|---|---| +| `src/showcase.ts` | One command: bare `nats-server` (JetStream for presence KV only — **no message streams, nothing persisted**), the observer, and the swarm. | +| `src/swarm.ts` | The load generator: *N* live-only endpoints (`channels:[]` pure publishers), sharded across `cluster` workers, each wired to a stable peer set (in-team peers + a cross-team bridge) it DMs and a team channel it posts to — that's what forms the lobe-and-bridge graph. | +| `src/observe.ts` | The scale-safe observer + SSE server (the bounded feed above). Runnable alone: `pnpm serve --space --server `. | +| `web/galaxy.html` + `web/galaxy.js` | The cosmos.gl GPU galaxy (loaded from a CDN — no bundler). | + +Live-only delivery throughout: messages are pure core-NATS pub/sub, so there's no JetStream +consumer-cardinality limit and nothing touches disk. diff --git a/examples/05-scale-showcase/package.json b/examples/05-scale-showcase/package.json new file mode 100644 index 00000000..0b26ede1 --- /dev/null +++ b/examples/05-scale-showcase/package.json @@ -0,0 +1,20 @@ +{ + "name": "@cotal-ai/example-05-scale-showcase", + "version": "0.0.0", + "private": true, + "license": "Apache-2.0", + "type": "module", + "scripts": { + "demo": "tsx src/showcase.ts", + "swarm": "tsx src/swarm.ts", + "serve": "tsx src/observe.ts", + "typecheck": "tsc -p tsconfig.json --noEmit", + "build": "tsc -p tsconfig.json" + }, + "dependencies": { + "@cotal-ai/core": "workspace:*" + }, + "devDependencies": { + "tsx": "^4.22.4" + } +} diff --git a/examples/05-scale-showcase/src/observe.ts b/examples/05-scale-showcase/src/observe.ts new file mode 100644 index 00000000..b1cec8a8 --- /dev/null +++ b/examples/05-scale-showcase/src/observe.ts @@ -0,0 +1,226 @@ +import { parseArgs } from "node:util"; +import { createServer, type ServerResponse } from "node:http"; +import { readFileSync } from "node:fs"; +import { fileURLToPath } from "node:url"; +import { dirname, join } from "node:path"; +import { + CotalEndpoint, + deliveryOf, + parseSubject, + spaceWildcard, + type CotalMessage, + type PresenceEvent, +} from "@cotal-ai/core"; + +/** + * The scale-safe observer for the galaxy. + * + * A read-only endpoint taps the whole space and folds every message into an in-memory graph + * model (agents, channel/role hubs, the edges between them). Crucially it NEVER emits one SSE + * event per message: it batches all structural deltas + a bounded "what lit up this frame" + * activity set and flushes them on a fixed ~15Hz tick. So the bytes pushed to the browser scale + * with the size of the GRAPH, not with the message rate — which is what makes 5k endpoints at + * thousands of msgs/sec survivable (the old per-message comet feed did not, and ate 50GB). + */ + +const here = dirname(fileURLToPath(import.meta.url)); +const WEB = join(here, "..", "web"); + +const TICK_MS = 66; // ~15Hz flush — the only cadence the browser ever sees + +type NodeKind = "agent" | "channel" | "role"; +interface GNode { id: string; kind: NodeKind; label: string; role?: string } + +interface Model { + nodes: GNode[]; + nodeIndex: Map; + links: [number, number][]; + linkIndex: Map; + status: Map; +} + +export interface ObserveOptions { server: string; space: string; port: number; creds?: string; monUrl?: string } + +export async function observe(opts: ObserveOptions): Promise { + const m: Model = { nodes: [], nodeIndex: new Map(), links: [], linkIndex: new Map(), status: new Map() }; + + // batched deltas, drained every tick + const pendNodes: number[] = []; + const pendLinks: number[] = []; + const pendStatus = new Map(); + const actNodes = new Set(); + const actLinks = new Set(); + let msgs = 0; // total real messages seen on the wire (chat/DM/anycast), the "delivered" counter + let mUni = 0, mMulti = 0, mAny = 0; // by delivery mode: unicast DM / multicast channel / anycast role + let latSum = 0, latN = 0; // rolling avg delivery latency: now − msg.ts (publish → observer receipt) + let bCpu = 0, bMem = 0, bConn = 0; // broker stats from NATS /varz monitoring + + const short = (id: string) => (id.length > 8 ? id.slice(0, 6) + "…" : id); + + function ensureNode(id: string, kind: NodeKind, label: string, role?: string): number { + let idx = m.nodeIndex.get(id); + if (idx === undefined) { + idx = m.nodes.length; + m.nodes.push({ id, kind, label, role }); + m.nodeIndex.set(id, idx); + pendNodes.push(idx); + } else if (role && !m.nodes[idx].role) { + m.nodes[idx].role = role; // late-learned role from a roster/message + } + return idx; + } + function ensureLink(a: number, b: number): number { + const key = a < b ? `${a}|${b}` : `${b}|${a}`; + let idx = m.linkIndex.get(key); + if (idx === undefined) { + idx = m.links.length; + m.links.push([a, b]); + m.linkIndex.set(key, idx); + pendLinks.push(idx); + } + return idx; + } + function setStatus(id: string, status: string): void { + if (m.status.get(id) === status) return; + m.status.set(id, status); + pendStatus.set(id, status); + } + + // ── the mesh observer (invisible to peers: no presence, no inbox) ── + const ep = new CotalEndpoint({ + space: opts.space, + servers: opts.server, + creds: opts.creds, + channels: [], + consume: false, + registerPresence: false, + watchPresence: true, + card: { name: "galaxy-observer", kind: "endpoint" }, + }); + ep.on("error", () => {}); + await ep.start(); + + // presence → nodes + statuses (so quiet agents still appear and color by status). Handle ONLY the + // changed peer per event — re-scanning the whole roster on every heartbeat is O(N) per beat and + // pegs the observer at scale (5k agents × heartbeats = millions of iterations/sec). + ep.on("presence", (e: PresenceEvent) => { + const p = e?.presence; if (!p?.card || p.card.kind === "endpoint") return; + const idx = ensureNode(p.card.id, "agent", p.card.name || short(p.card.id), p.card.role); + actNodes.add(idx); // a heartbeat is a faint sign of life + setStatus(p.card.id, p.status); + }); + + // every comm → fold into the model + mark what lit up (no per-message emit) + ep.tap((subject: string, msg: CotalMessage | undefined) => { + const mode = deliveryOf(subject); + if (!mode || !msg) return; + const parsed = parseSubject(subject); + if (!parsed) return; + msgs++; + if (mode === "unicast") mUni++; else if (mode === "chat") mMulti++; else mAny++; + if (msg.ts) { latSum += Math.max(0, Date.now() - msg.ts); latN++; } + const sender = parsed.sender; + const from = ensureNode(sender, "agent", msg.from?.name || short(sender), msg.from?.role); + actNodes.add(from); + + if (mode === "unicast") { + // a DM is a real agent↔agent communication edge — peer-to-peer structure + const to = ensureNode(parsed.rest, "agent", short(parsed.rest)); + actNodes.add(to); + actLinks.add(ensureLink(from, to)); + } else if (mode === "chat") { + // a channel post wires the sender to a shared channel HUB — teams gather into lobes around these + const ch = ensureNode(`#${parsed.rest}`, "channel", `#${parsed.rest}`); + actNodes.add(ch); + actLinks.add(ensureLink(from, ch)); + } + // anycast stays ambient: the sender already pulsed above, no node/edge. + }, { subject: spaceWildcard(opts.space) }); + + // ── SSE fan-out, fixed cadence ── + const clients = new Set(); + const send = (res: ServerResponse, event: string, data: unknown) => { + if (!res.writableEnded) res.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`); + }; + const broadcast = (event: string, data: unknown) => { for (const res of clients) send(res, event, data); }; + + setInterval(() => { + if (pendNodes.length) { broadcast("nodes", pendNodes.map((i) => m.nodes[i])); pendNodes.length = 0; } + if (pendLinks.length) { broadcast("links", pendLinks.map((i) => m.links[i])); pendLinks.length = 0; } + if (pendStatus.size) { broadcast("status", Object.fromEntries(pendStatus)); pendStatus.clear(); } + if (actNodes.size || actLinks.size) { + broadcast("act", { p: [...actNodes], l: [...actLinks] }); + actNodes.clear(); actLinks.clear(); + } + }, TICK_MS); + + // broker cpu/mem/conns from NATS' native HTTP monitoring (/varz) — polled on its own cadence + const pollVarz = async (): Promise => { + if (!opts.monUrl) return; + try { + const r = await fetch(`${opts.monUrl}/varz`); + if (!r.ok) return; + const v = await r.json() as { cpu?: number; mem?: number; connections?: number }; + bCpu = Math.round(v.cpu ?? 0); bMem = Math.round((v.mem ?? 0) / 1e6); bConn = v.connections ?? 0; + } catch { /* monitoring not up yet */ } + }; + if (opts.monUrl) { void pollVarz(); setInterval(() => void pollVarz(), 1000); } + + // the perf/meta feed (counter + per-mode rates + avg latency + broker stats), on its own slow cadence + let lastMsgs = 0, lUni = 0, lMulti = 0, lAny = 0, lastT = Date.now(); + const meta = () => { + const now = Date.now(), dt = (now - lastT) / 1000; + const per = (cur: number, last: number) => (dt > 0 ? Math.round((cur - last) / dt) : 0); + const rate = per(msgs, lastMsgs), dm = per(mUni, lUni), chan = per(mMulti, lMulti), any = per(mAny, lAny); + const lat = latN ? Math.round((latSum / latN) * 100) / 100 : 0; + lastMsgs = msgs; lUni = mUni; lMulti = mMulti; lAny = mAny; lastT = now; latSum = 0; latN = 0; + return { msgs, rate, dm, chan, any, lat, cpu: bCpu, mem: bMem, conns: bConn }; + }; + setInterval(() => broadcast("meta", meta()), 500); + + // ── http: static page + the SSE feed + a stats probe ── + const server = createServer((req, res) => { + const path = (req.url ?? "/").split("?")[0]; + if (path === "/feed") { + res.writeHead(200, { "content-type": "text/event-stream", "cache-control": "no-cache", connection: "keep-alive" }); + clients.add(res); + // full snapshot first; the client dedups by id so an overlapping tick delta is harmless + send(res, "snapshot", { nodes: m.nodes, links: m.links, status: Object.fromEntries(m.status) }); + send(res, "meta", { msgs, rate: 0, dm: 0, chan: 0, any: 0, lat: 0, cpu: bCpu, mem: bMem, conns: bConn }); + req.on("close", () => clients.delete(res)); + return; + } + if (path === "/api/stats") { + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify({ space: opts.space, nodes: m.nodes.length, links: m.links.length, clients: clients.size })); + return; + } + const file = path === "/" ? "galaxy.html" : path.slice(1); + try { + const body = readFileSync(join(WEB, file)); // read FIRST — a miss throws before any header is written + const type = file.endsWith(".js") ? "text/javascript; charset=utf-8" : "text/html; charset=utf-8"; + res.writeHead(200, { "content-type": type, "cache-control": "no-cache" }); + res.end(body); + } catch { + res.writeHead(404).end("not found"); + } + }); + await new Promise((ready) => server.listen(opts.port, "127.0.0.1", ready)); + console.log(`galaxy observer → http://127.0.0.1:${opts.port}/ (space ${opts.space})`); + + const stop = async () => { for (const r of clients) r.end(); server.close(); await ep.stop().catch(() => {}); process.exit(0); }; + process.on("SIGINT", stop); process.on("SIGTERM", stop); +} + +// allow running standalone: `tsx src/observe.ts --space bench --server nats://127.0.0.1:4711 --port 7900` +if (import.meta.url === `file://${process.argv[1]}`) { + const { values } = parseArgs({ args: process.argv.slice(2).filter((a) => a !== "--"), options: { + server: { type: "string" }, space: { type: "string" }, port: { type: "string" }, creds: { type: "string" }, + } }); + void observe({ + server: values.server ?? "nats://127.0.0.1:4711", + space: values.space ?? "bench", + port: values.port ? Number(values.port) : 7900, + creds: values.creds ? readFileSync(values.creds, "utf8") : undefined, + }); +} diff --git a/examples/05-scale-showcase/src/showcase.ts b/examples/05-scale-showcase/src/showcase.ts new file mode 100644 index 00000000..f732aecb --- /dev/null +++ b/examples/05-scale-showcase/src/showcase.ts @@ -0,0 +1,64 @@ +import { parseArgs } from "node:util"; +import { spawn } from "node:child_process"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join, dirname } from "node:path"; +import { fileURLToPath } from "node:url"; +import { isReachable } from "@cotal-ai/core"; +import { observe } from "./observe.js"; + +/** + * `pnpm demo --n 5000` — the whole showcase in one command: + * 1. a bare nats-server (JetStream on for presence KV, but NO message streams → zero persistence) + * 2. the scale-safe observer + galaxy web page + * 3. the swarm of N real endpoints, all communicating + * Then prints the URL. Live-only, nothing touches disk except the throwaway store. Ctrl-C tears it all down. + */ + +const here = dirname(fileURLToPath(import.meta.url)); + +async function main(): Promise { + const { values } = parseArgs({ args: process.argv.slice(2).filter((a) => a !== "--"), options: { + n: { type: "string" }, workers: { type: "string" }, rate: { type: "string" }, conc: { type: "string" }, + heartbeat: { type: "string" }, port: { type: "string" }, web: { type: "string" }, space: { type: "string" }, + } }); + const n = values.n ? Number(values.n) : 2000; + const workers = values.workers ? Number(values.workers) : (n > 2500 ? Math.min(8, Math.ceil(n / 1500)) : 1); + const rate = values.rate ? Number(values.rate) : 0.2; // per-agent msg rate (the observer handles it fine with a single viewer) + const heartbeat = values.heartbeat ? Number(values.heartbeat) : (n >= 12000 ? 12000 : 4000); // longer heartbeats at huge n → far less presence churn on the broker + observer + // keep total concurrent connects (conc × workers) well under the OS listen backlog (somaxconn ~128) + const conc = values.conc ? Number(values.conc) : Math.max(16, Math.floor(256 / workers)); + const brokerPort = values.port ? Number(values.port) : 4733; + const monPort = brokerPort + 1; // NATS HTTP monitoring (/varz) — the observer polls it for broker cpu/mem + const webPort = values.web ? Number(values.web) : 7900; + const space = values.space ?? "galaxy"; + const server = `nats://127.0.0.1:${brokerPort}`; + + // 1) bare broker — JetStream for presence KV only; no chat/dm/task streams are created, so messages + // are pure core pub/sub (fire-and-forget). Nothing is persisted. `-m` enables HTTP monitoring. + const store = mkdtempSync(join(tmpdir(), "cotal-galaxy-")); + const broker = spawn("nats-server", ["-js", "-sd", store, "-p", String(brokerPort), "-m", String(monPort), "-a", "127.0.0.1"], { stdio: "ignore" }); + broker.on("error", (e: NodeJS.ErrnoException) => { + console.error(e.code === "ENOENT" ? "✗ nats-server not found on PATH (brew install nats-server)" : `✗ broker: ${e.message}`); + process.exit(1); + }); + for (let i = 0; i < 50 && !(await isReachable(server)); i++) await new Promise((r) => setTimeout(r, 200)); + if (!(await isReachable(server))) { console.error(`✗ broker never came up at ${server}`); process.exit(1); } + console.log(`✓ broker at ${server} (live-only, no persistence)`); + + // 2) observer + galaxy page (in-process) + await observe({ server, space, port: webPort, monUrl: `http://127.0.0.1:${monPort}` }); + + // 3) swarm as a child process (keeps cluster sharding isolated from this orchestrator) + const swarm = spawn(process.execPath, [ + ...process.execArgv, join(here, "swarm.ts"), + "--server", server, "--space", space, "--n", String(n), "--workers", String(workers), "--rate", String(rate), "--conc", String(conc), "--heartbeat", String(heartbeat), + ], { stdio: "inherit" }); + + console.log(`\n ★ ${n} endpoints (${workers} worker${workers > 1 ? "s" : ""}) → open http://127.0.0.1:${webPort}/\n`); + + const stop = () => { try { swarm.kill("SIGTERM"); } catch {} try { broker.kill("SIGTERM"); } catch {} rmSync(store, { recursive: true, force: true }); process.exit(0); }; + process.on("SIGINT", stop); process.on("SIGTERM", stop); +} + +void main(); diff --git a/examples/05-scale-showcase/src/swarm.ts b/examples/05-scale-showcase/src/swarm.ts new file mode 100644 index 00000000..1953b2d6 --- /dev/null +++ b/examples/05-scale-showcase/src/swarm.ts @@ -0,0 +1,143 @@ +import { parseArgs } from "node:util"; +import cluster from "node:cluster"; +import { CotalEndpoint } from "@cotal-ai/core"; + +/** + * The load generator: N real Cotal endpoints, live-only (no persistence), all talking. + * + * The graph is the REAL peer graph — agents connect to other AGENTS they actually message, not to + * abstract channel hubs. Each agent has a stable set of collaborators (a handful inside its team + + * a couple cross-team), and it DMs them; the observer turns those DMs into agent↔agent edges. That + * yields a connected small-world network with team communities and bridges — a graph agents could + * really have — instead of disconnected hub-and-spoke stars. + * + * Endpoints use deterministic ids (`a`) so any worker can address any peer by id even + * when sharded across processes (unicast just needs the target id; it's fire-and-forget on the wire). + */ + +const ROLES = ["planner", "builder", "reviewer", "researcher", "ops"]; +const TEAM_SIZE = 100; // agents per team community +const IN_PEERS = 5; // collaborators inside the team +const CROSS = 1; // structured cross-team bridge on each bridge agent (target in another team) +const BRIDGE_EVERY = 3; // ~1 in 3 agents bridges to another team (~1.6k bridges) — interwoven, not islands +const RANDOM_EVERY = 15; // ~1 in 15 also gets a FULLY-random long-range link (no team structure) — the + // long arcs that make a small-world net read organic rather than engineered + +const rng = (seed: number) => { let s = seed >>> 0; return () => { s = (Math.imul(s, 1597334677) + 1) >>> 0; return s / 4294967296; }; }; + +/** Stable peer set for agent gi (by global id), mostly in-team + a cross-team bridge. */ +function peersFor(gi: number, total: number): string[] { + const team = Math.floor(gi / TEAM_SIZE); + const start = team * TEAM_SIZE, end = Math.min(total, start + TEAM_SIZE); + const r = rng(gi + 1), peers = new Set(); + for (let k = 0; k < IN_PEERS && end - start > 1; k++) { + const p = start + Math.floor(r() * (end - start)); + if (p !== gi) peers.add(`a${p}`); + } + const bridges = gi % BRIDGE_EVERY === 0 ? CROSS : 0; + for (let k = 0; k < bridges; k++) { + let p = Math.floor(r() * total); + if (Math.floor(p / TEAM_SIZE) === team) p = (p + TEAM_SIZE) % total; + if (p !== gi) peers.add(`a${p}`); + } + if (gi % RANDOM_EVERY === 0) { const p = Math.floor(r() * total); if (p !== gi) peers.add(`a${p}`); } // random long-range arc + return [...peers]; +} + +export interface SwarmOptions { + server: string; space: string; n: number; base: number; total: number; rate: number; heartbeatMs: number; conc: number; +} + +export async function runSwarm(o: SwarmOptions): Promise { + const eps: { ep: CotalEndpoint; peers: string[]; team: number }[] = []; + let connected = 0, failed = 0, sent = 0, errSamples = 0; + + const make = async (gi: number): Promise => { + const peers = peersFor(gi, o.total); + const team = Math.floor(gi / TEAM_SIZE); + let lastErr = ""; + // Retry transient connect failures: under a large connect burst the broker's accept backlog + // (macOS somaxconn defaults to 128) drops connections; a couple of backed-off retries recover them. + for (let attempt = 0; attempt < 6; attempt++) { + const ep = new CotalEndpoint({ + space: o.space, servers: o.server, + card: { id: `a${gi}`, name: `bot-${gi}`, role: ROLES[gi % ROLES.length], kind: "agent" }, + channels: [], consume: false, registerPresence: true, watchPresence: false, + heartbeatMs: o.heartbeatMs, + }); + ep.on("error", () => {}); + try { + await ep.start(); + for (const p of peers) ep.unicast(p, "hi").catch(() => {}); // seed each peer edge (fire-and-forget) + ep.multicast("hello", { channel: `team${team}` }).catch(() => {}); // seed this agent's channel-hub edge + eps.push({ ep, peers, team }); + connected++; + return; + } catch (e) { + lastErr = (e as Error)?.message ?? String(e); + await ep.stop().catch(() => {}); + if (attempt < 5) await new Promise((r) => setTimeout(r, 200 * (attempt + 1))); + } + } + failed++; + if (errSamples < 6) { errSamples++; console.error(`${tag}connect FAILED (gi=${gi}) after retries: ${lastErr}`); } + }; + + const tag = cluster.worker ? `w${cluster.worker.id} ` : ""; + const t0 = Date.now(); + for (let i = 0; i < o.n; i += o.conc) { + await Promise.all(Array.from({ length: Math.min(o.conc, o.n - i) }, (_, k) => make(o.base + i + k))); + } + console.log(`${tag}connected ${connected}/${o.n} (${failed} failed) in ${((Date.now() - t0) / 1000).toFixed(1)}s`); + + // one scheduler emits a slice of the target rate each tick — not N timers + const tickMs = 100; + const perTick = Math.max(1, Math.round(o.n * o.rate * (tickMs / 1000))); + const pick = () => eps[(Math.random() * eps.length) | 0]; + setInterval(() => { + for (let k = 0; k < perTick && eps.length; k++) { + const me = pick(); if (!me) continue; + const roll = Math.random(); + if (roll < 0.7 && me.peers.length) me.ep.unicast(me.peers[(Math.random() * me.peers.length) | 0], "ping").catch(() => {}); // pulse a real edge + else if (roll < 0.9) me.ep.multicast(`m${sent}`, { channel: `team${me.team}` }).catch(() => {}); // ambient channel chatter + else me.ep.anycast(ROLES[(Math.random() * ROLES.length) | 0], "task").catch(() => {}); + sent++; + } + }, tickMs); + setInterval(() => console.log(`${tag}live=${eps.length} sent=${sent} rss=${(process.memoryUsage().rss / 1e9).toFixed(2)}GB`), 5000); + + process.on("SIGINT", () => process.exit(0)); + process.on("SIGTERM", () => process.exit(0)); +} + +/** Run N endpoints, sharded across `workers` child processes (cluster). With workers=1, in-process. */ +export async function startSwarm(opts: Omit & { total: number; workers: number }): Promise { + const { total, workers, ...rest } = opts; + if (workers > 1 && cluster.isPrimary) { + const per = Math.ceil(total / workers); + // Don't override argv — workers inherit the full flags and read their slice from COTAL_SHARD. + for (let w = 0; w < workers; w++) { + const base = w * per, count = Math.min(per, total - base); + if (count > 0) cluster.fork({ COTAL_SHARD: JSON.stringify({ base, count }) }); + } + return; + } + const shard = process.env.COTAL_SHARD ? JSON.parse(process.env.COTAL_SHARD) as { base: number; count: number } : { base: 0, count: total }; + await runSwarm({ ...rest, total, n: shard.count, base: shard.base }); +} + +if (import.meta.url === `file://${process.argv[1]}` || process.env.COTAL_SHARD) { + const { values } = parseArgs({ args: process.argv.slice(2).filter((a) => a !== "--"), options: { + server: { type: "string" }, space: { type: "string" }, n: { type: "string" }, + workers: { type: "string" }, rate: { type: "string" }, heartbeat: { type: "string" }, conc: { type: "string" }, + } }); + void startSwarm({ + server: values.server ?? "nats://127.0.0.1:4711", + space: values.space ?? "bench", + total: values.n ? Number(values.n) : 1000, + workers: values.workers ? Number(values.workers) : 1, + rate: values.rate ? Number(values.rate) : 0.2, + heartbeatMs: values.heartbeat ? Number(values.heartbeat) : 4000, + conc: values.conc ? Number(values.conc) : 150, + }); +} diff --git a/examples/05-scale-showcase/tsconfig.json b/examples/05-scale-showcase/tsconfig.json new file mode 100644 index 00000000..051d08e2 --- /dev/null +++ b/examples/05-scale-showcase/tsconfig.json @@ -0,0 +1,8 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { + "rootDir": "./src", + "outDir": "./dist" + }, + "include": ["src"] +} diff --git a/examples/05-scale-showcase/web/galaxy.html b/examples/05-scale-showcase/web/galaxy.html new file mode 100644 index 00000000..5fa2487e --- /dev/null +++ b/examples/05-scale-showcase/web/galaxy.html @@ -0,0 +1,59 @@ + + + + + + Cotal · galaxy + + + +
+
+
+
0 agents
+
live on the mesh
+
+
0
+
messages delivered · 0/s
+
DM 0 · channel 0 · anycast 0 /s
+
avg delivery 0 ms
+
+
broker 0% cpu · 0 MB
+
0 conns · 0 edges · 0 fps
+
+
Cotal
a live agent mesh
+
+
colour = role · node size = connections · large nodes = channel hubs
+
planner builder reviewer researcher ops channel · warm lines = bridges
+
hover a node for its role & peers · click to isolate · click away to release
+
+ +
+ + + diff --git a/examples/05-scale-showcase/web/galaxy.js b/examples/05-scale-showcase/web/galaxy.js new file mode 100644 index 00000000..89b998d2 --- /dev/null +++ b/examples/05-scale-showcase/web/galaxy.js @@ -0,0 +1,212 @@ +// Cotal galaxy — GPU mesh visualization (cosmos.gl). +// +// Why this stays light where the old 2D graph blew up: every buffer is FIXED-SIZE and reused. +// Pulses are not objects — an SSE "act" frame just sets heat=1 on a few indices; the render loop +// decays heat and writes one node-color + one node-size + one link-color buffer per frame. So +// per-frame work scales with what's ON SCREEN (nodes+links), never with the message rate. The +// force layout itself runs entirely on the GPU. +import { Graph } from "https://esm.sh/@cosmos.gl/graph@3.0.0"; + +const stage = document.getElementById("stage"); +const tip = document.getElementById("tip"); + +// Colour = ROLE (the agent's type). Five colour-blind-safe Okabe-Ito hues; channel hubs are cyan. +const TEAM_SIZE = 100; // mirrors swarm.ts: agent a is in team floor(gi/TEAM_SIZE) — used to detect cross-team bridges +const ROLE = { + planner: [0.0, 0.447, 0.698], // #0072B2 blue + builder: [0.0, 0.620, 0.451], // #009E73 green + reviewer: [0.941, 0.894, 0.259], // #F0E442 yellow + researcher: [0.8, 0.475, 0.655], // #CC79A7 purple + ops: [0.835, 0.369, 0.0], // #D55E00 vermillion +}; +const CHANNEL = [0.43, 0.86, 0.95]; // bright cyan — channel hubs +const UNKNOWN = [0.6, 0.66, 0.71]; +const baseRGB = (n) => (n.kind === "channel" ? CHANNEL : (ROLE[n.role] || UNKNOWN)); +function communityOf(n) { // team index, only for detecting cross-team bridge edges (warm-coloured) + if (n.kind === "channel") { const m = /team(\d+)/.exec(n.label || n.id); return m ? +m[1] : -1; } + const m = /^a(\d+)$/.exec(n.id); return m ? Math.floor(+m[1] / TEAM_SIZE) : -1; +} + +// model +const idIndex = new Map(); const nodes = []; +const linkSet = new Set(); const links = []; +const status = new Map(); +let dirty = false; + +function addNode(n) { + const i = idIndex.get(n.id); + if (i !== undefined) { if (n.role && !nodes[i].role) { nodes[i].role = n.role; dirty = true; } return; } + idIndex.set(n.id, nodes.length); nodes.push(n); dirty = true; +} +function addLink(a, b) { const k = a < b ? a + "|" + b : b + "|" + a; if (linkSet.has(k)) return; linkSet.add(k); links.push([a, b]); dirty = true; } +function setStat(id, s) { if (status.get(id) !== s) { status.set(id, s); dirty = true; } } // only dirty on a real change + +// gpu buffers +let positions = new Float32Array(0), base = new Float32Array(0), baseSize = new Float32Array(0), psize = new Float32Array(0), colors = new Float32Array(0); +let linkArr = new Float32Array(0), linkBase = new Float32Array(0), linkCol = new Float32Array(0), linkW = new Float32Array(0), linkKind = new Uint8Array(0); +let heat = new Float32Array(0), heatL = new Float32Array(0); +let degArr = new Int32Array(0), commArr = new Int32Array(0); +const posById = new Map(); + +// click-to-isolate: when a node is focused, these hold the survivors; everything else dims. +let focusNodes = null, focusLinks = null; + +// "messages delivered" odometer: server gives an authoritative total + rate every 0.5s; we count +// up smoothly between updates (rate × dt) so the number visibly streams instead of jumping. +let msgTotal = 0, msgRate = 0, msgShown = 0, lastMsgT = performance.now(); +let perf = { lat: 0, cpu: 0, mem: 0, conns: 0, dm: 0, chan: 0, any: 0 }; // latency + broker stats + per-mode rates + +const graph = new Graph(stage, { + backgroundColor: "#0a0e18", // lifted off harsh black; the #vignette overlay darkens the edges for depth + spaceSize: 8192, + pointDefaultColor: "#9aa6b5", linkDefaultColor: "#33405a", + pointSizeScale: 1.0, scalePointsOnZoom: true, linkWidthScale: 1.0, + // A real org: channel hubs gather each team into a tight lobe, DM springs bind collaborators, + // repulsion pushes teams apart so the ~50 lobes separate, gravity holds the whole galaxy centered + // and compact (so it stays framed). It SETTLES and stops — no drift; the staggered fits below frame + // it as it expands and onSimulationEnd frames the final extent. + simulationGravity: 0.045, simulationCenter: 0, simulationRepulsion: 2.8, simulationRepulsionTheta: 1.5, + simulationLinkSpring: 0.5, simulationLinkDistance: 55, simulationFriction: 0.85, simulationDecay: 6000, + fitViewOnInit: false, + onSimulationEnd: () => { if (!userControl && !focusNodes) graph.fitView(700, 0.18, false); settled = true; }, // layout reached its extent → stop auto-framing + // discovery without UI tax: hover for a tooltip, click a node to isolate it + its edges. + renderHoveredPointRing: true, hoveredPointRingColor: "#eaf1ff", hoveredPointCursor: "pointer", + onPointMouseOver: (i) => showTip(i), + onPointMouseOut: () => hideTip(), + onPointClick: (i) => setFocus(i), + onBackgroundClick: () => clearFocus(), +}); + +function setFocus(i) { + const nbrs = graph.getNeighboringPointIndices(i) || []; + focusNodes = new Set([i, ...nbrs]); + focusLinks = new Set(graph.getConnectedLinkIndices(i) || []); +} +function clearFocus() { if (focusNodes) { focusNodes = null; focusLinks = null; graph.fitView(700, 0.18, false); } } + +// auto-frame the galaxy as it assembles & settles — but the MOMENT the viewer touches the view +// (scroll/drag), hand over control PERMANENTLY so we never yank their zoom/pan again. (The earlier +// bug: streaming nodes kept re-enabling auto-fit, so it fought the user's zoom.) +let userControl = false, settled = false; +const takeControl = () => { userControl = true; }; +["wheel", "pointerdown"].forEach((ev) => stage.addEventListener(ev, takeControl, { passive: true })); +// Re-frame ONLY while the layout is still settling, and NEVER run physics during the camera move +// (enableSimulation=false). Otherwise framing perpetually re-heats the sim and nothing ever stops. +setInterval(() => { if (!userControl && !focusNodes && !settled) graph.fitView(600, 0.18, false); }, 1500); + +// tooltip follows the cursor (robust to whatever event type cosmos hands the callback) +let mx = 0, my = 0; +addEventListener("mousemove", (e) => { mx = e.clientX; my = e.clientY; if (tip.style.opacity === "1") place(); }); +function place() { tip.style.left = Math.min(mx + 14, innerWidth - 200) + "px"; tip.style.top = (my + 14) + "px"; } +function showTip(i) { + const n = nodes[i]; if (!n) return; + const deg = degArr[i] || 0, c = commArr[i]; + tip.innerHTML = n.kind === "channel" + ? `${n.label}
team channel · ${deg} member${deg === 1 ? "" : "s"}` + : `${n.label}
${c >= 0 ? "team " + c + " · " : ""}${n.role || "agent"} · ${deg} peer${deg === 1 ? "" : "s"}`; + tip.style.opacity = "1"; place(); +} +function hideTip() { tip.style.opacity = "0"; } + +function rebuild() { + const cur = graph.getPointPositions(); + if (cur && cur.length) for (let i = 0; i < nodes.length && i * 2 + 1 < cur.length; i++) posById.set(nodes[i].id, [cur[2 * i], cur[2 * i + 1]]); + const N = nodes.length, L = links.length; + positions = new Float32Array(2 * N); base = new Float32Array(4 * N); baseSize = new Float32Array(N); psize = new Float32Array(N); colors = new Float32Array(4 * N); + const nh = new Float32Array(N); nh.set(heat.subarray(0, Math.min(heat.length, N))); heat = nh; + const deg = new Int32Array(N), comm = new Int32Array(N); // degree → size; community → colour + for (let i = 0; i < L; i++) { deg[links[i][0]]++; deg[links[i][1]]++; } + degArr = deg; commArr = comm; + for (let i = 0; i < N; i++) { + const n = nodes[i]; comm[i] = communityOf(n); + const p = posById.get(n.id) || [(Math.random() - 0.5) * 1400, (Math.random() - 0.5) * 1400]; + positions[2 * i] = p[0]; positions[2 * i + 1] = p[1]; + const c = baseRGB(n); // colour by role (agents) / cyan (channel hubs) + const off = status.get(n.id) === "offline", o = 4 * i; + base[o] = c[0]; base[o + 1] = c[1]; base[o + 2] = c[2]; base[o + 3] = off ? 0.22 : 1; + baseSize[i] = n.kind === "channel" ? 6 + Math.min(14, Math.sqrt(deg[i]) * 1.3) : 2 + Math.min(7, Math.sqrt(deg[i]) * 1.1); + } + linkArr = new Float32Array(2 * L); linkBase = new Float32Array(4 * L); linkCol = new Float32Array(4 * L); linkW = new Float32Array(L); linkKind = new Uint8Array(L); + const nhl = new Float32Array(L); nhl.set(heatL.subarray(0, Math.min(heatL.length, L))); heatL = nhl; + for (let i = 0; i < L; i++) { + const a = links[i][0], b = links[i][1], o = 4 * i; linkArr[2 * i] = a; linkArr[2 * i + 1] = b; + const isChan = nodes[a].kind === "channel" || nodes[b].kind === "channel"; + const ca = comm[a], cb = comm[b]; + if (isChan) { + // channel-membership spoke: soft cyan, draws the lobe around its hub + linkKind[i] = 1; linkBase[o] = 0.30; linkBase[o + 1] = 0.72; linkBase[o + 2] = 0.82; linkBase[o + 3] = 0.13; linkW[i] = 0.6; + } else if (ca !== cb && ca >= 0 && cb >= 0) { + // cross-team BRIDGE — the small-world backbone, WARM amber over the cool communities: glows + // across the dark gaps, instantly legible, and gorgeous against the nebula palette + linkKind[i] = 2; linkBase[o] = 1.0; linkBase[o + 1] = 0.72; linkBase[o + 2] = 0.34; linkBase[o + 3] = 0.26; linkW[i] = 1.2; + } else { + // in-team DM: dim, faded by connectivity so degree-1 stragglers vanish + const md = Math.min(deg[a], deg[b]); const alpha = md <= 1 ? 0.05 : Math.min(0.26, 0.09 + md * 0.012); + linkBase[o] = 0.55; linkBase[o + 1] = 0.62; linkBase[o + 2] = 0.85; linkBase[o + 3] = alpha; linkW[i] = md <= 1 ? 0.5 : 0.7; + } + } + graph.setPointPositions(positions); graph.setPointSizes(baseSize); graph.setLinks(linkArr); graph.setLinkWidths(linkW); + graph.start(0.6); settled = false; // new data → re-layout; allow re-framing until it settles again +} + +// render loop — bounded, one color + one size upload per frame +let fps = 0, fa = 0, ft = performance.now(); +function frame() { + if (dirty) { dirty = false; rebuild(); } + const N = heat.length, L = heatL.length, fa_ = focusNodes; + for (let i = 0; i < N; i++) { + heat[i] *= 0.94; const h = heat[i], k = h * 0.9, o = 4 * i; // linger longer so more lit at once + colors[o] = base[o] + (1 - base[o]) * k; colors[o + 1] = base[o + 1] + (1 - base[o + 1]) * k; + colors[o + 2] = base[o + 2] + (1 - base[o + 2]) * k; colors[o + 3] = Math.min(1, base[o + 3] + h * 0.6); + psize[i] = baseSize[i] * (1 + h * 0.7); // talking nodes swell, then relax + if (fa_ && !fa_.has(i)) { colors[o] *= 0.3; colors[o + 1] *= 0.3; colors[o + 2] *= 0.3; colors[o + 3] = 0.08; } + } + for (let i = 0; i < L; i++) { + heatL[i] *= 0.965; const h = heatL[i], o = 4 * i; // edges glow ~3× longer → reads alive at 20k density + linkCol[o] = linkBase[o] + (1 - linkBase[o]) * h; linkCol[o + 1] = linkBase[o + 1] + (1 - linkBase[o + 1]) * h; + linkCol[o + 2] = linkBase[o + 2] + (1 - linkBase[o + 2]) * h; + linkCol[o + 3] = Math.min(0.95, linkBase[o + 3] + h * (linkKind[i] === 1 ? 0.85 : 0.75)); + if (focusLinks) linkCol[o + 3] = focusLinks.has(i) ? Math.min(0.95, 0.5 + h * 0.5) : 0.02; + } + if (N) { graph.setPointColors(colors); graph.setPointSizes(psize); } + if (L) graph.setLinkColors(linkCol); + graph.render(); + // smooth message odometer + const now = performance.now(), dt = (now - lastMsgT) / 1000; lastMsgT = now; + msgShown = Math.max(msgShown + msgRate * dt, msgShown); + if (msgShown < msgTotal) msgShown = Math.min(msgTotal, msgShown + (msgTotal - msgShown) * 0.2); + fa++; + if (now - ft >= 250) { fps = Math.round((fa * 1000) / (now - ft)); fa = 0; ft = now; hud(); } + requestAnimationFrame(frame); +} +function hud() { + document.getElementById("nodes").textContent = nodes.reduce((a, n) => a + (n.kind === "agent" ? 1 : 0), 0).toLocaleString(); + document.getElementById("links").textContent = links.length.toLocaleString(); + document.getElementById("fps").textContent = fps; + document.getElementById("msgs").textContent = Math.floor(msgShown).toLocaleString(); + document.getElementById("rate").textContent = msgRate.toLocaleString(); + document.getElementById("lat").textContent = perf.lat < 10 ? perf.lat.toFixed(2) : Math.round(perf.lat); + document.getElementById("dm").textContent = perf.dm.toLocaleString(); + document.getElementById("chan").textContent = perf.chan.toLocaleString(); + document.getElementById("any").textContent = perf.any.toLocaleString(); + document.getElementById("bcpu").textContent = perf.cpu; + document.getElementById("bmem").textContent = perf.mem.toLocaleString(); + document.getElementById("bconn").textContent = perf.conns.toLocaleString(); +} + +// SSE +const es = new EventSource("/feed"); +// snapshot is re-sent on every SSE reconnect — addNode/addLink/setStat dedup, so a reconnect that +// changes nothing leaves dirty=false and triggers NO rebuild (the old unconditional dirty=true here +// re-built the whole 20k/123k graph on each reconnect → multi-second freeze → missed SSE → reconnect loop). +es.addEventListener("snapshot", (e) => { const d = JSON.parse(e.data); for (const n of d.nodes) addNode(n); for (const l of d.links) addLink(l[0], l[1]); for (const [id, s] of Object.entries(d.status || {})) setStat(id, s); }); +es.addEventListener("nodes", (e) => { for (const n of JSON.parse(e.data)) addNode(n); }); +es.addEventListener("links", (e) => { for (const l of JSON.parse(e.data)) addLink(l[0], l[1]); }); +es.addEventListener("status", (e) => { for (const [id, s] of Object.entries(JSON.parse(e.data))) setStat(id, s); }); +es.addEventListener("act", (e) => { const d = JSON.parse(e.data); for (const i of d.p) if (i < heat.length) heat[i] = 1; for (const i of d.l) if (i < heatL.length) heatL[i] = 1; }); +es.addEventListener("meta", (e) => { const d = JSON.parse(e.data); if (d.msgs >= msgTotal) msgTotal = d.msgs; msgRate = d.rate; if (msgShown === 0) msgShown = d.msgs; perf = { lat: d.lat ?? 0, cpu: d.cpu ?? 0, mem: d.mem ?? 0, conns: d.conns ?? 0, dm: d.dm ?? 0, chan: d.chan ?? 0, any: d.any ?? 0 }; }); + +addEventListener("resize", () => graph.render()); +// press "h" to hide all chrome (HUD/legend/title) for a clean hero capture +addEventListener("keydown", (e) => { if (e.key === "h" || e.key === "H") document.body.classList.toggle("chrome-off"); }); +requestAnimationFrame(frame); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 70f74e8c..142dd1e3 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -117,6 +117,16 @@ importers: specifier: ^4.22.4 version: 4.22.4 + examples/05-scale-showcase: + dependencies: + '@cotal-ai/core': + specifier: workspace:* + version: link:../../packages/core + devDependencies: + tsx: + specifier: ^4.22.4 + version: 4.22.4 + extensions/cmux: dependencies: '@cotal-ai/core':