diff --git a/docs/groupmind-local.md b/docs/groupmind-local.md new file mode 100644 index 0000000..5617a59 --- /dev/null +++ b/docs/groupmind-local.md @@ -0,0 +1,52 @@ +# GroupMind Local — offline coordination relay + +The fleet normally coordinates through cloud GroupMind. When the internet is +down but the machines are still on the same LAN (everyone home, ISP outage), +there is no hub and the agents go silent. **GroupMind Local** is a small +always-on service — meant to run on the machine that never sleeps (e.g. a Mac +mini) — that speaks the subset of the GroupMind message API the fleet uses, so +any agent can fail over to it and keep talking with **zero internet**. + +Local models on each machine (an on-device LLM per box) + GroupMind Local as the +hub = a fleet that reasons and coordinates fully offline. + +## Run it + +```bash +# defaults: :18790, store ~/.iak/local-relay.jsonl, LAN-open +npm run relay + +# or explicitly +node scripts/local-relay.mjs --port 18790 --store ~/.iak/local-relay.jsonl --token "$TOKEN" +``` + +Config via env: `IAK_RELAY_PORT`, `IAK_RELAY_STORE`, `IAK_RELAY_TOKEN`. +If `IAK_RELAY_TOKEN` is set, every request must present it via `X-API-Key` or +`Authorization: Bearer`. Unset means LAN-open — fine on a trusted home LAN only. + +## API (GroupMind-compatible subset) + +| Method | Path | Body / Query | Notes | +| --- | --- | --- | --- | +| `GET` | `/health` | — | always open; reports message count | +| `POST` | `/api/v1/messages` | `{ room, body, from?, metadata? }` | create a message | +| `POST` | `/api/v1/rooms/:room/messages` | `{ body, from?, metadata? }` | create (room in path) | +| `GET` | `/api/v1/rooms/:room/messages` | `?limit=&since=` | newest-first, GroupMind shape | + +An agent points its room client at `http://:18790/api/v1` when cloud +GroupMind is unreachable — same paths, just a different base URL. + +## Scope (MVP) and what's next + +This is a **LAN message store + serve** — deliberately not all of GroupMind +(no auth DB, no realtime, no reactions). Every stored message carries a UUID and +`origin: "local-relay"`. + +Not yet built (the deliberate next PRs): + +1. **Client failover** — the IAK room client trying cloud first, then the local + relay, automatically. +2. **Sync-on-reconnect** — merging the relay's offline history back into cloud + GroupMind when the internet returns, deduped by message id. This is the hard + part; the UUID + `origin` on every message exist so it can merge without + reordering or duplicates. diff --git a/package.json b/package.json index c912f11..2520146 100644 --- a/package.json +++ b/package.json @@ -10,7 +10,8 @@ "scripts": { "test": "node --test test/*.test.mjs packages/user-intent-kit/test/*.test.js", "start": "node bin/cli.mjs serve", - "mcp": "node bin/iak-mcp.mjs" + "mcp": "node bin/iak-mcp.mjs", + "relay": "node scripts/local-relay.mjs" }, "keywords": [ "ide", diff --git a/scripts/local-relay.mjs b/scripts/local-relay.mjs new file mode 100644 index 0000000..ef73265 --- /dev/null +++ b/scripts/local-relay.mjs @@ -0,0 +1,192 @@ +#!/usr/bin/env node +// SPDX-License-Identifier: AGPL-3.0-only +// +// iak-local-relay — IAK's offline coordination hub. +// +// The fleet normally coordinates through cloud GroupMind (Vercel). When the +// internet is down but the machines are still on the same LAN (e.g. everyone +// home, ISP outage), there is no hub and the agents go silent. This is a small +// always-on service — meant to run on the Mac mini, the machine that never +// sleeps — that speaks the SUBSET of GroupMind's message API the fleet uses, so +// an agent can fail over to `http://:/api/v1` and keep +// talking with zero internet. +// +// Scope (MVP): a LAN message store + serve. It is NOT the whole of GroupMind +// (no auth DB, no realtime, no reactions) and it does NOT yet reconcile its +// offline history back into cloud GroupMind when the internet returns — that +// sync-on-reconnect step (merge by message id) is the deliberate next PR. Every +// message it stores carries a UUID + `origin: "local-relay"` so that future +// sync can dedup and merge without reordering. +// +// Zero runtime deps (built-in http/fs/crypto only), matching the rest of IAK. +// +// Routes: +// GET /health +// POST /api/v1/messages { room, body, from?, metadata? } +// POST /api/v1/rooms/:room/messages { body, from?, metadata? } +// GET /api/v1/rooms/:room/messages?limit=&since= +// +// Auth: if IAK_RELAY_TOKEN is set, every request must present it via X-API-Key +// or `Authorization: Bearer`. If unset, the relay is LAN-open (it logs one +// warning) — acceptable on a trusted home LAN, not on an untrusted network. +// +// Config (env or startRelay() args): IAK_RELAY_PORT (default 18790), +// IAK_RELAY_STORE (default ~/.iak/local-relay.jsonl), IAK_RELAY_TOKEN. + +import http from 'node:http'; +import { appendFileSync, mkdirSync, readFileSync, existsSync } from 'node:fs'; +import { randomUUID } from 'node:crypto'; +import { homedir } from 'node:os'; +import { dirname, join } from 'node:path'; + +const DEFAULT_PORT = 18790; +const DEFAULT_STORE = join(homedir(), '.iak', 'local-relay.jsonl'); +const MAX_LIMIT = 200; +const DEFAULT_LIMIT = 50; +const MAX_BODY_BYTES = 256 * 1024; // reject absurd payloads + +function readMessages(storePath) { + if (!existsSync(storePath)) return []; + const out = []; + for (const line of readFileSync(storePath, 'utf8').split('\n')) { + if (!line.trim()) continue; + try { out.push(JSON.parse(line)); } catch { /* skip a corrupt line, never crash a read */ } + } + return out; +} + +function appendMessage(storePath, msg) { + mkdirSync(dirname(storePath), { recursive: true }); + appendFileSync(storePath, JSON.stringify(msg) + '\n'); +} + +function send(res, status, obj) { + const payload = JSON.stringify(obj); + res.writeHead(status, { 'Content-Type': 'application/json', 'Content-Length': Buffer.byteLength(payload) }); + res.end(payload); +} + +function readJsonBody(req) { + return new Promise((resolve, reject) => { + let size = 0; + const chunks = []; + req.on('data', (c) => { + size += c.length; + if (size > MAX_BODY_BYTES) { reject(new Error('payload too large')); req.destroy(); return; } + chunks.push(c); + }); + req.on('end', () => { + const raw = Buffer.concat(chunks).toString('utf8'); + if (!raw.trim()) return resolve({}); + try { resolve(JSON.parse(raw)); } catch { reject(new Error('invalid JSON')); } + }); + req.on('error', reject); + }); +} + +// Build a stored message from an incoming create request. Returns { msg } or +// { error } for a validation failure. +function buildMessage({ room, body, from, metadata }) { + if (!room || typeof room !== 'string') return { error: 'room is required' }; + if (!body || typeof body !== 'string') return { error: 'body is required' }; + return { + msg: { + id: randomUUID(), + room, + from: typeof from === 'string' && from ? from : 'unknown', + body, + metadata: metadata ?? null, + created_at: new Date().toISOString(), + origin: 'local-relay', + }, + }; +} + +/** + * Start the relay. Returns the http.Server (already listening). + * @param {{port?:number, storePath?:string, token?:string, logger?:(m:string)=>void}} opts + */ +export function startRelay(opts = {}) { + const envPort = Number(process.env.IAK_RELAY_PORT); + const port = opts.port ?? (Number.isFinite(envPort) ? envPort : DEFAULT_PORT); + const storePath = opts.storePath ?? process.env.IAK_RELAY_STORE ?? DEFAULT_STORE; + const token = opts.token ?? process.env.IAK_RELAY_TOKEN ?? ''; + const log = opts.logger ?? ((m) => process.stderr.write(`[iak-local-relay] ${m}\n`)); + + if (!token) log('WARNING: no IAK_RELAY_TOKEN set — relay is LAN-open (fine on a trusted home LAN only).'); + + function authorized(req) { + if (!token) return true; + const bearer = (req.headers['authorization'] || '').replace(/^Bearer\s+/i, ''); + return req.headers['x-api-key'] === token || bearer === token; + } + + const server = http.createServer(async (req, res) => { + try { + const url = new URL(req.url, `http://localhost:${port}`); + const path = url.pathname.replace(/\/+$/, '') || '/'; + + if (req.method === 'GET' && path === '/health') { + return send(res, 200, { ok: true, service: 'iak-local-relay', messages: readMessages(storePath).length }); + } + + if (!authorized(req)) return send(res, 401, { error: 'unauthorized' }); + + // GET /api/v1/rooms/:room/messages + const getMatch = path.match(/^\/api\/v1\/rooms\/([^/]+)\/messages$/); + if (req.method === 'GET' && getMatch) { + const room = decodeURIComponent(getMatch[1]); + const limit = Math.min(Math.max(1, Number(url.searchParams.get('limit')) || DEFAULT_LIMIT), MAX_LIMIT); + const since = url.searchParams.get('since'); + let msgs = readMessages(storePath).filter((m) => m.room === room); + if (since) msgs = msgs.filter((m) => m.created_at > since); + // JSONL append order IS chronological, so it orders even messages that + // share a millisecond timestamp. Take the last `limit` (newest) and + // reverse to newest-first, matching GroupMind's shape. + const messages = msgs.slice(-limit).reverse(); + return send(res, 200, { messages, count: messages.length }); + } + + // POST /api/v1/messages { room, body, ... } + if (req.method === 'POST' && path === '/api/v1/messages') { + const b = await readJsonBody(req); + const { msg, error } = buildMessage({ room: b.room, body: b.body, from: b.from, metadata: b.metadata }); + if (error) return send(res, 400, { error }); + appendMessage(storePath, msg); + return send(res, 201, msg); + } + + // POST /api/v1/rooms/:room/messages { body, ... } + const postMatch = path.match(/^\/api\/v1\/rooms\/([^/]+)\/messages$/); + if (req.method === 'POST' && postMatch) { + const room = decodeURIComponent(postMatch[1]); + const b = await readJsonBody(req); + const { msg, error } = buildMessage({ room, body: b.body, from: b.from, metadata: b.metadata }); + if (error) return send(res, 400, { error }); + appendMessage(storePath, msg); + return send(res, 201, msg); + } + + return send(res, 404, { error: 'not found' }); + } catch (err) { + const status = /payload too large/.test(err?.message) ? 413 : /invalid JSON/.test(err?.message) ? 400 : 500; + return send(res, status, { error: err?.message || 'internal error' }); + } + }); + + server.listen(port, () => log(`listening on :${port} store=${storePath}`)); + return server; +} + +// CLI entry +if (import.meta.url === `file://${process.argv[1]}`) { + const arg = (name) => { + const i = process.argv.indexOf(name); + return i !== -1 ? process.argv[i + 1] : undefined; + }; + startRelay({ + port: Number(arg('--port')) || undefined, + storePath: arg('--store'), + token: arg('--token'), + }); +} diff --git a/test/local-relay.test.mjs b/test/local-relay.test.mjs new file mode 100644 index 0000000..9dc29ea --- /dev/null +++ b/test/local-relay.test.mjs @@ -0,0 +1,128 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { once } from 'node:events'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { rmSync } from 'node:fs'; +import { randomUUID } from 'node:crypto'; +import { startRelay } from '../scripts/local-relay.mjs'; + +// Start a relay on an ephemeral port with an isolated store, run fn, tear down. +async function withRelay(opts, fn) { + const storePath = join(tmpdir(), `iak-relay-test-${randomUUID()}.jsonl`); + const server = startRelay({ port: 0, storePath, logger: () => {}, ...opts }); + await once(server, 'listening'); + const port = server.address().port; + const base = `http://127.0.0.1:${port}`; + try { + return await fn({ base, storePath, headers: opts.token ? { 'X-API-Key': opts.token } : {} }); + } finally { + server.close(); + await once(server, 'close'); + try { rmSync(storePath, { force: true }); } catch { /* ignore */ } + } +} + +const j = (h = {}) => ({ 'Content-Type': 'application/json', ...h }); + +test('health reports ok and message count', async () => { + await withRelay({}, async ({ base }) => { + const r = await fetch(`${base}/health`); + assert.equal(r.status, 200); + const d = await r.json(); + assert.equal(d.ok, true); + assert.equal(d.service, 'iak-local-relay'); + assert.equal(d.messages, 0); + }); +}); + +test('POST /api/v1/messages stores and GET returns it (round trip)', async () => { + await withRelay({}, async ({ base, headers }) => { + const post = await fetch(`${base}/api/v1/messages`, { + method: 'POST', headers: j(headers), + body: JSON.stringify({ room: 'r1', body: 'hello', from: '@mm' }), + }); + assert.equal(post.status, 201); + const msg = await post.json(); + assert.ok(msg.id, 'has id'); + assert.equal(msg.room, 'r1'); + assert.equal(msg.from, '@mm'); + assert.equal(msg.body, 'hello'); + assert.equal(msg.origin, 'local-relay'); + assert.ok(msg.created_at, 'has created_at'); + + const get = await fetch(`${base}/api/v1/rooms/r1/messages`, { headers }); + assert.equal(get.status, 200); + const { messages, count } = await get.json(); + assert.equal(count, 1); + assert.equal(messages[0].body, 'hello'); + assert.equal(messages[0].id, msg.id); + }); +}); + +test('POST /api/v1/rooms/:room/messages (path form) works and room filters', async () => { + await withRelay({}, async ({ base, headers }) => { + await fetch(`${base}/api/v1/rooms/alpha/messages`, { + method: 'POST', headers: j(headers), body: JSON.stringify({ body: 'in-alpha' }), + }); + await fetch(`${base}/api/v1/messages`, { + method: 'POST', headers: j(headers), body: JSON.stringify({ room: 'beta', body: 'in-beta' }), + }); + const alpha = await (await fetch(`${base}/api/v1/rooms/alpha/messages`, { headers })).json(); + assert.equal(alpha.count, 1); + assert.equal(alpha.messages[0].body, 'in-alpha'); + const beta = await (await fetch(`${base}/api/v1/rooms/beta/messages`, { headers })).json(); + assert.equal(beta.count, 1); + assert.equal(beta.messages[0].body, 'in-beta'); + }); +}); + +test('GET honors limit and returns newest-first', async () => { + await withRelay({}, async ({ base, headers }) => { + for (const n of [1, 2, 3]) { + await fetch(`${base}/api/v1/messages`, { + method: 'POST', headers: j(headers), body: JSON.stringify({ room: 'r', body: `m${n}` }), + }); + } + const { messages, count } = await (await fetch(`${base}/api/v1/rooms/r/messages?limit=2`, { headers })).json(); + assert.equal(count, 2); + assert.equal(messages[0].body, 'm3', 'newest first'); + }); +}); + +test('POST without body is rejected 400', async () => { + await withRelay({}, async ({ base, headers }) => { + const r = await fetch(`${base}/api/v1/messages`, { + method: 'POST', headers: j(headers), body: JSON.stringify({ room: 'r' }), + }); + assert.equal(r.status, 400); + }); +}); + +test('malformed JSON body is rejected 400 (never crashes)', async () => { + await withRelay({}, async ({ base, headers }) => { + const r = await fetch(`${base}/api/v1/messages`, { method: 'POST', headers: j(headers), body: 'not json' }); + assert.equal(r.status, 400); + }); +}); + +test('token auth: rejects missing/wrong key, accepts correct', async () => { + await withRelay({ token: 'secret123' }, async ({ base }) => { + const noKey = await fetch(`${base}/api/v1/rooms/r/messages`); + assert.equal(noKey.status, 401); + const wrong = await fetch(`${base}/api/v1/rooms/r/messages`, { headers: { 'X-API-Key': 'nope' } }); + assert.equal(wrong.status, 401); + const right = await fetch(`${base}/api/v1/rooms/r/messages`, { headers: { 'X-API-Key': 'secret123' } }); + assert.equal(right.status, 200); + // health stays open even with a token configured + const health = await fetch(`${base}/health`); + assert.equal(health.status, 200); + }); +}); + +test('unknown route is 404', async () => { + await withRelay({}, async ({ base, headers }) => { + const r = await fetch(`${base}/api/v1/nope`, { headers }); + assert.equal(r.status, 404); + }); +});