// Relate hub — the control plane. One hub, many spaces: // * settings UI (public/hub/) + JSON API under /hub/api (see docs/PLATFORM.md) // * space registry: spaces//config.json (user config, secrets inside) — spaces/ is gitignored // * supervisor: runs each space as its own `node server.js` worker (own port, own data dir) // * recording router: watches one Plaud inbox, classifies each recording, forwards it to the right space(s) // The hub never imports the engine (lib/config.js) — workers are separate processes. import http from 'node:http'; import { spawn } from 'node:child_process'; import { readFileSync, writeFileSync, existsSync, mkdirSync, readdirSync, rmSync, appendFileSync, statSync, renameSync } from 'node:fs'; import { join, dirname, extname } from 'node:path'; import { fileURLToPath } from 'node:url'; import { randomBytes, scryptSync, timingSafeEqual, createHmac } from 'node:crypto'; import net from 'node:net'; import { connectorTypes, describeTypes } from './lib/connectors/index.js'; import { validateConfig, withDefaults } from './lib/connectors/connector.js'; const ROOT = dirname(fileURLToPath(import.meta.url)); const HOME = process.env.RELATE_HOME || join(ROOT, 'spaces'); const PORT = Number(process.env.HUB_PORT || 5080); // ports browsers/fetch refuse to connect to (WHATWG "bad ports") — e.g. 5060/5061 are SIP const UNSAFE_PORTS = new Set([1, 7, 9, 11, 13, 15, 17, 19, 20, 21, 22, 23, 25, 37, 42, 43, 53, 69, 77, 79, 87, 95, 101, 102, 103, 104, 109, 110, 111, 113, 115, 117, 119, 123, 135, 137, 139, 143, 161, 179, 389, 427, 465, 512, 513, 514, 515, 526, 530, 531, 532, 540, 548, 554, 556, 563, 587, 601, 636, 989, 990, 993, 995, 1719, 1720, 1723, 2049, 3659, 4045, 4190, 5060, 5061, 6000, 6566, 6665, 6666, 6667, 6668, 6669, 6679, 6697, 10080]); const HOST = process.env.HUB_HOST || '127.0.0.1'; const SETTINGS = join(HOME, 'hub.json'); const ROUTER_DIR = join(HOME, '_router'); const VERSION = JSON.parse(readFileSync(join(ROOT, 'package.json'), 'utf8')).version || '0.0.0'; const KEEP = '__set__'; // placeholder the UI sends back for "keep the stored secret" mkdirSync(HOME, { recursive: true }); const log = (...a) => console.log(new Date().toISOString().slice(11, 19), ...a); // ---------------- templates ---------------- const TEMPLATES = [ { id: 'relationship', label: 'Relationship', description: 'One important person. Tracks mood, needs, sensitive topics, promises and who owes a reply.', defaults: { acceptGroups: false }, plugins: ['occasions', 'reply-nudge'] }, { id: 'business', label: 'Business', description: 'Your business chats and groups. Open loops, who is waiting on you, commitments, blockers.', defaults: { acceptGroups: true } }, { id: 'team', label: 'Team', description: 'A team or project. Owners, decisions, unclaimed work, blockers and collaboration health.', defaults: { acceptGroups: true } }, ]; // ---------------- settings & auth ---------------- function readJson(f, dflt) { try { return JSON.parse(readFileSync(f, 'utf8')); } catch { return dflt; } } function writeJson(f, v) { mkdirSync(dirname(f), { recursive: true }); const tmp = f + '.tmp'; writeFileSync(tmp, JSON.stringify(v, null, 2)); renameSync(tmp, f); } let settings = readJson(SETTINGS, {}); settings.secret ||= randomBytes(24).toString('hex'); settings.ai ||= { base: '', key: '', model: '', maxCallsPerHour: 40 }; settings.episodeRouter ||= { enabled: false, inbox: '', minConfidence: 0.6, speakerMap: [] }; writeJson(SETTINGS, settings); const hashPw = (pw, salt = randomBytes(16).toString('hex')) => `${salt}:${scryptSync(String(pw), salt, 32).toString('hex')}`; function checkPw(pw) { if (!settings.passwordHash) return false; const [salt, h] = settings.passwordHash.split(':'); const a = Buffer.from(scryptSync(String(pw), salt, 32).toString('hex')), b = Buffer.from(h); return a.length === b.length && timingSafeEqual(a, b); } const sessionToken = () => createHmac('sha256', settings.secret).update('hub:' + settings.passwordHash).digest('hex'); function cookies(req) { const o = {}; for (const p of String(req.headers.cookie || '').split(';')) { const i = p.indexOf('='); if (i > 0) o[p.slice(0, i).trim()] = decodeURIComponent(p.slice(i + 1).trim()); } return o; } const authed = (req) => !!settings.passwordHash && cookies(req).relate_hub === sessionToken(); const setCookie = (res) => res.setHeader('Set-Cookie', `relate_hub=${sessionToken()}; Path=/; Max-Age=5184000; HttpOnly; SameSite=Lax`); // ---------------- space registry ---------------- const spaceDir = (id) => join(HOME, id); const cfgFile = (id) => join(spaceDir(id), 'config.json'); const validId = (id) => /^[a-z0-9][a-z0-9-]{0,39}$/.test(id || ''); function listIds() { return readdirSync(HOME, { withFileTypes: true }).filter((d) => d.isDirectory() && !d.name.startsWith('_') && existsSync(cfgFile(d.name))).map((d) => d.name).sort(); } const loadCfg = (id) => readJson(cfgFile(id), null); const saveCfg = (id, c) => writeJson(cfgFile(id), c); const token = (n = 24) => randomBytes(n).toString('hex'); function slug(name) { return String(name || 'space').toLowerCase().normalize('NFKD').replace(/[^a-z0-9]+/g, '-').replace(/^-+|-+$/g, '').slice(0, 30) || 'space'; } async function freePort(start = 5081) { const used = new Set(listIds().map((id) => loadCfg(id)?.port).filter(Boolean)); for (let p = start; p < start + 200; p++) { if (used.has(p) || p === PORT || UNSAFE_PORTS.has(p)) continue; const bindable = await new Promise((r) => { const s = net.createServer().once('error', () => r(false)).once('listening', () => s.close(() => r(true))).listen(p, '127.0.0.1'); }); // Windows lets us bind 127.0.0.1 even while another process holds the port on all interfaces, so also try to connect const taken = bindable && await new Promise((r) => { const c = net.connect(p, '127.0.0.1').once('connect', () => { c.destroy(); r(true); }).once('error', () => r(false)); c.setTimeout(800, () => { c.destroy(); r(false); }); }); if (bindable && !taken) return p; } throw new Error('no free port'); } // secret handling: which fields are secrets, mask them on read, keep stored values on "__set__" const TOP_SECRETS = [['ai', 'key'], ['auth', 'password'], ['auth', 'secret'], ['auth', 'apiToken']]; async function connSecretKeys(type) { const T = (await connectorTypes()).get(type); return (T?.configSchema || []).filter((f) => f.type === 'secret').map((f) => f.key); } // plugin settings secrets come from each plugin's configSchema (read via the sandboxed inspection, cached by mtime) const inspectCache = new Map(); // file -> { mtime, result } async function pluginSecretKeys(spaceId, name) { const f = [join(spacePlugins(spaceId), `${name}.js`), join(SHARED_PLUGINS, `${name}.js`), join(BUILTIN_PLUGINS, `${name}.js`)].find((x) => existsSync(x)); if (!f) return []; const mtime = statSync(f).mtimeMs, hit = inspectCache.get(f); const r = hit && hit.mtime === mtime ? hit.result : await inspectPlugin(f, name); inspectCache.set(f, { mtime, result: r }); return (r.meta?.configSchema || []).filter((x) => x.type === 'secret').map((x) => x.key); } async function maskPluginSettings(spaceId, name, settings) { const out = { ...(settings || {}) }; for (const k of await pluginSecretKeys(spaceId, name)) if (k in out) out[k] = out[k] ? KEEP : ''; return out; } async function maskCfg(c) { const out = structuredClone(c); for (const [a, b] of TOP_SECRETS) if (out[a] && b in out[a]) out[a][b] = out[a][b] ? KEEP : ''; for (const conn of out.connectors || []) for (const k of await connSecretKeys(conn.type)) if (k in conn) conn[k] = conn[k] ? KEEP : ''; for (const name of Object.keys(out.plugins || {})) out.plugins[name] = await maskPluginSettings(c.id, name, out.plugins[name]); return out; } async function unmask(incoming, stored) { const out = structuredClone(incoming); for (const [a, b] of TOP_SECRETS) if (out[a]?.[b] === KEEP) out[a][b] = stored?.[a]?.[b] || ''; for (const conn of out.connectors || []) { const prev = (stored?.connectors || []).find((x) => x.id === conn.id); for (const k of await connSecretKeys(conn.type)) if (conn[k] === KEEP) conn[k] = prev?.[k] || ''; } for (const [name, p] of Object.entries(out.plugins || {})) if (p && typeof p === 'object') for (const [k, v] of Object.entries(p)) if (v === KEEP) p[k] = stored?.plugins?.[name]?.[k] || ''; return out; } function deepMerge(base, patch) { if (Array.isArray(patch) || typeof patch !== 'object' || patch === null) return patch; const out = { ...(base && typeof base === 'object' && !Array.isArray(base) ? base : {}) }; for (const [k, v] of Object.entries(patch)) out[k] = deepMerge(out[k], v); return out; } async function validateSpace(c) { const errors = []; if (!c.name) errors.push('Name is required'); if (!TEMPLATES.some((t) => t.id === c.template)) errors.push(`Unknown template "${c.template}"`); const types = await connectorTypes(); const ids = new Set(); for (const conn of c.connectors || []) { if (!conn.id || !/^[\w-]{1,40}$/.test(conn.id)) { errors.push(`Connector id "${conn.id}" is invalid`); continue; } if (conn.id === '_space') { errors.push('Connector id "_space" is reserved'); continue; } if (ids.has(conn.id)) errors.push(`Duplicate connector id "${conn.id}"`); ids.add(conn.id); const T = types.get(conn.type); if (!T) { errors.push(`Connector ${conn.id}: unknown type "${conn.type}"`); continue; } if (conn.enabled !== false) for (const e of validateConfig(T.configSchema, withDefaults(T.configSchema, conn))) errors.push(`Connector ${conn.id}: ${e}`); } for (const ch of c.chats || []) if (!ids.has(ch.connector)) errors.push(`Chat "${ch.name || ch.chatId}" refers to missing connector "${ch.connector}"`); if (c.template === 'relationship' && c.target?.chatId && !(c.chats || []).some((ch) => ch.chatId === c.target.chatId)) errors.push('The partner chat must be in the chat list'); return errors; } // session id the engine stores a chat under (mirrors lib/sources.js sessionIdFor) async function sessionFor(c, connectorId, chatId) { const conn = (c.connectors || []).find((x) => x.id === connectorId); const T = conn && (await connectorTypes()).get(conn.type); const platform = conn?.type === 'ingest' ? (conn.platform || 'custom') : (T?.platform || 'custom'); return platform === 'wechat' ? chatId : `${platform}:${chatId}`; } // runtime.json = what the worker actually runs: config + global AI defaults + hub-owned plumbing async function writeRuntime(id) { const c = loadCfg(id); const rt = structuredClone(c); rt.id = id; rt.contextType = c.template; rt.dataDir = join(spaceDir(id), 'data'); rt.pluginDir = join(spaceDir(id), 'plugins'); const g = settings.ai || {}; rt.ai = { ...(c.ai || {}) }; for (const k of ['base', 'key', 'model']) if (!rt.ai[k]) rt.ai[k] = g[k] || ''; if (!rt.ai.maxCallsPerHour) rt.ai.maxCallsPerHour = g.maxCallsPerHour || 40; rt.ai.enabled = rt.ai.enabled !== false && !!(rt.ai.base && rt.ai.key && rt.ai.model); rt.auth = { ...(c.auth || {}) }; rt.auth.enabled = !!rt.auth.password; rt.connectors = [...(c.connectors || [])]; // every space gets a hub-owned HTTP ingest endpoint (/api/ingest/_space, space ingest token). The recording // router delivers through it; recordings need no allowlist, chat messages still only land for listed chats if (c.ingest?.token) rt.connectors = [...rt.connectors.filter((x) => x.id !== '_space'), { id: '_space', type: 'ingest', token: c.ingest.token, platform: 'custom' }]; rt.chats = await Promise.all((c.chats || []).map(async (ch) => ({ ...ch, session: await sessionFor(c, ch.connector, ch.chatId) }))); if (c.target?.chatId) rt.target = { ...c.target, session: await sessionFor(c, c.target.connector || c.chats?.find((x) => x.chatId === c.target.chatId)?.connector, c.target.chatId) }; mkdirSync(rt.dataDir, { recursive: true }); mkdirSync(rt.pluginDir, { recursive: true }); const f = join(spaceDir(id), 'runtime.json'); writeJson(f, rt); return f; } // ---------------- supervisor ---------------- const procs = new Map(); // id -> { child, status, startedAt, lastError, lines:[], restarts:[] } const rec = (id) => procs.get(id) || procs.set(id, { status: 'stopped', lines: [], restarts: [] }).get(id); function pushLine(id, line) { const r = rec(id); r.lines.push(line); if (r.lines.length > 800) r.lines.splice(0, r.lines.length - 800); try { appendFileSync(join(spaceDir(id), 'worker.log'), line + '\n'); } catch {} } async function workerFetch(id, path, opts = {}) { const c = loadCfg(id); const headers = { ...(opts.headers || {}) }; if (c?.auth?.apiToken && !headers.Authorization) headers.Authorization = `Bearer ${c.auth.apiToken}`; // callers (ingest) may bring their own return fetch(`http://127.0.0.1:${c.port}${path}`, { ...opts, headers, signal: AbortSignal.timeout(opts.timeout || 8000) }); } // is THIS space's worker answering on its port? (checks the id, so a stranger on the port is never adopted) async function portAnswers(id) { try { const r = await workerFetch(id, '/api/space', { timeout: 1500 }); return r.ok && (await r.json()).id === id; } catch { return false; } } async function startSpace(id) { const c = loadCfg(id); if (!c) throw new Error('no such space'); const r = rec(id); if (r.child && (r.status === 'running' || r.status === 'starting')) return r.status; c.auth ||= {}; c.auth.apiToken ||= token(); c.auth.secret ||= token(); c.ingest ||= {}; c.ingest.token ||= token(); saveCfg(id, c); if (await portAnswers(id)) { r.status = 'running'; r.adopted = true; pushLine(id, `[hub] port ${c.port} already serving this space — adopted`); return r.status; } const runtime = await writeRuntime(id); const args = []; if (settings.sandbox) args.push('--permission', `--allow-fs-read=${ROOT}`, `--allow-fs-read=${spaceDir(id)}`, `--allow-fs-write=${spaceDir(id)}`); args.push(join(ROOT, 'server.js')); const child = spawn(process.execPath, args, { cwd: ROOT, env: { ...process.env, RELATE_CONFIG: runtime, PORT: String(c.port) }, windowsHide: true, stdio: ['ignore', 'pipe', 'pipe'] }); Object.assign(r, { child, status: 'starting', startedAt: Date.now(), lastError: null, adopted: false, stopping: false }); pushLine(id, `[hub] starting worker pid ${child.pid} on :${c.port}`); const onData = (buf) => { for (const l of String(buf).split(/\r?\n/)) if (l.trim()) { pushLine(id, l); if (/Error|ERR_|unhandled/i.test(l)) r.lastError = l.slice(0, 300); } }; child.stdout.on('data', onData); child.stderr.on('data', onData); child.on('exit', (code) => { const was = r.stopping; r.child = null; r.status = was ? 'stopped' : 'crashed'; pushLine(id, `[hub] worker exited (code ${code})${was ? '' : ' — unexpected'}`); if (!was) { // crash: restart with backoff, at most 5 times in 10 minutes const now = Date.now(); r.restarts = r.restarts.filter((t) => now - t < 600000); if (r.restarts.length < 5) { r.restarts.push(now); const d = 2000 * 2 ** (r.restarts.length - 1); pushLine(id, `[hub] restarting in ${d / 1000}s`); setTimeout(() => { if (rec(id).status === 'crashed') startSpace(id).catch(() => {}); }, d); } else pushLine(id, '[hub] crashed 5 times in 10 minutes — giving up until you start it again'); } }); (async () => { // readiness: wait for the port to answer, then deliver any queued recordings for (let i = 0; i < 120 && r.child === child; i++) { if (await portAnswers(id)) { r.status = 'running'; pushLine(id, '[hub] worker ready'); flushOutbox(id); return; } await new Promise((ok) => setTimeout(ok, 500)); } if (r.child === child && r.status === 'starting') { r.lastError = 'worker did not become ready within 60s'; pushLine(id, `[hub] ${r.lastError}`); } })(); c.autostart = true; saveCfg(id, c); return r.status; } async function stopSpace(id, { remember = true } = {}) { const r = rec(id); if (remember) { const c = loadCfg(id); if (c) { c.autostart = false; saveCfg(id, c); } } if (r.child) { r.stopping = true; r.child.kill(); await new Promise((ok) => { const t = setTimeout(ok, 5000); r.child?.once('exit', () => { clearTimeout(t); ok(); }); }); } else if (r.adopted) pushLine(id, '[hub] this worker was started outside the hub; stop it there'); r.status = r.child ? r.status : 'stopped'; return r.status; } async function spaceSummary(id) { const c = loadCfg(id), r = rec(id); return { id, name: c.name, template: c.template, port: c.port, url: `http://127.0.0.1:${c.port}/`, status: r.status, pid: r.child?.pid || null, startedAt: r.startedAt || null, lastError: r.lastError || null, connectors: (c.connectors || []).map((x) => ({ id: x.id, type: x.type, enabled: x.enabled !== false })), chatCount: (c.chats || []).length, }; } // ---------------- connector test / chat picker (without saving) ---------------- async function resolveConnCfg({ type, config = {}, space }) { const cfg = { ...config }; const stored = (space && loadCfg(space)?.connectors || []).find((x) => x.id === cfg.id) || listIds().flatMap((i) => loadCfg(i)?.connectors || []).find((x) => x.id === cfg.id && x.type === type); for (const k of await connSecretKeys(type)) if (cfg[k] === KEEP) cfg[k] = stored?.[k] || ''; return cfg; } async function withTempConnector(body, fn) { const T = (await connectorTypes()).get(body.type); if (!T) return { ok: false, error: `unknown connector type "${body.type}"` }; const cfg = withDefaults(T.configSchema, await resolveConnCfg(body)); cfg.id ||= 'test'; const errs = validateConfig(T.configSchema, cfg); if (errs.length) return { ok: false, error: errs.join('; ') }; const dataDir = body.space && validId(body.space) ? join(spaceDir(body.space), 'data') : join(HOME, '_scratch'); mkdirSync(dataDir, { recursive: true }); const conn = T.create(cfg, { onMessage: () => false, onEpisode: () => false, onStatus: () => {}, log, dataDir }); try { return await fn(conn); } finally { try { conn.stop?.(); } catch {} } } // ---------------- AI (global test + recording classifier) ---------------- async function aiChat({ base, key, model }, messages, { json = false, maxTokens = 600, timeout = 60000 } = {}) { const body = { model, messages, max_tokens: maxTokens, temperature: 0.2 }; if (json) body.response_format = { type: 'json_object' }; const r = await fetch(String(base).replace(/\/+$/, '') + '/chat/completions', { method: 'POST', headers: { Authorization: `Bearer ${key}`, 'Content-Type': 'application/json' }, body: JSON.stringify(body), signal: AbortSignal.timeout(timeout) }); const txt = await r.text(); if (!r.ok) throw new Error(`HTTP ${r.status}: ${txt.slice(0, 160)}`); return JSON.parse(txt).choices?.[0]?.message?.content || ''; } function looseJson(t) { try { return JSON.parse(t); } catch {} const s = t.indexOf('{'), e = t.lastIndexOf('}'); if (s >= 0 && e > s) { try { return JSON.parse(t.slice(s, e + 1)); } catch {} } return null; } // ---------------- recording router (Plaud inbox → spaces) ---------------- const EP_DIR = join(ROUTER_DIR, 'episodes'); mkdirSync(EP_DIR, { recursive: true }); const epFile = (id) => join(EP_DIR, `${String(id).replace(/[^\w.-]/g, '_')}.json`); let routerConn = null; async function classify(ep) { const ai = settings.ai || {}; const spaces = listIds().map((id) => { const c = loadCfg(id); return { id, name: c.name, template: c.template, about: c.description || '', people: (c.chats || []).map((x) => x.name).slice(0, 20) }; }); if (!spaces.length) return []; if (!(ai.base && ai.key && ai.model)) return null; // no AI → manual review const excerpt = ep.turns.slice(0, 60).map((t) => `${t.speaker}: ${t.text}`).join('\n').slice(0, 5000); const raw = await aiChat(ai, [ { role: 'system', content: 'You route a recorded conversation to the spaces (areas of the user\'s life/work) it is relevant to. A recording may belong to several spaces or none. Output only JSON.' }, { role: 'user', content: `Spaces:\n${JSON.stringify(spaces)}\n\nRecording "${ep.title}" (${new Date(ep.ts).toISOString()}), participants: ${ep.participants.join(', ')}\n${ep.summary ? 'Summary: ' + ep.summary + '\n' : ''}Transcript excerpt:\n${excerpt}\n\nReturn {"scores":[{"space":"","score":0..1,"why":"short"}]} for every space.` }, ], { json: true, maxTokens: 500 }); const p = looseJson(raw); return (p?.scores || []).filter((s) => spaces.some((x) => x.id === s.space)).map((s) => ({ space: s.space, score: Math.max(0, Math.min(1, Number(s.score) || 0)), why: String(s.why || '').slice(0, 120) })).sort((a, b) => b.score - a.score); } function onRouterEpisode(ep) { if (existsSync(epFile(ep.id))) return false; writeJson(epFile(ep.id), { episode: ep, routing: { status: 'classifying', suggested: [], routedTo: [], pending: [] } }); log(`router: new recording "${ep.title}"`); (async () => { const doc = readJson(epFile(ep.id)); try { const scores = await classify(ep); doc.routing.suggested = scores || []; const min = Number(settings.episodeRouter?.minConfidence) || 0.6; const pick = (scores || []).filter((s) => s.score >= min).map((s) => s.space); if (scores && pick.length) { doc.routing.status = 'routed'; doc.routing.pending = pick; } else doc.routing.status = 'review'; } catch (e) { doc.routing.status = 'review'; doc.routing.error = e.message; } writeJson(epFile(ep.id), doc); for (const s of doc.routing.pending) flushOutbox(s); })(); return true; } // deliver queued recordings to a space (called on route, and whenever the space becomes ready) async function flushOutbox(spaceId) { if (rec(spaceId).status !== 'running') return; const c = loadCfg(spaceId); if (!c?.ingest?.token) return; for (const f of readdirSync(EP_DIR).filter((x) => x.endsWith('.json'))) { const doc = readJson(join(EP_DIR, f)); if (!doc?.routing?.pending?.includes(spaceId)) continue; try { const r = await workerFetch(spaceId, '/api/ingest/_space', { method: 'POST', headers: { Authorization: `Bearer ${c.ingest.token}`, 'Content-Type': 'application/json' }, body: JSON.stringify({ episodes: [{ ...doc.episode, tags: [spaceId] }] }) }); if (!r.ok) throw new Error(`HTTP ${r.status}`); doc.routing.pending = doc.routing.pending.filter((x) => x !== spaceId); doc.routing.routedTo = [...new Set([...doc.routing.routedTo, spaceId])]; writeJson(join(EP_DIR, f), doc); log(`router: "${doc.episode.title}" → ${spaceId}`); } catch (e) { log(`router: delivery to ${spaceId} failed (${e.message}) — will retry when it restarts`); } } } async function startRouter() { try { routerConn?.stop(); } catch {} routerConn = null; const R = settings.episodeRouter || {}; if (!R.enabled || !R.inbox) return; const T = (await connectorTypes()).get('plaud'); if (!T) { log('router: plaud connector type unavailable'); return; } routerConn = T.create(withDefaults(T.configSchema, { id: 'router', inbox: R.inbox, speakerMap: R.speakerMap || [], pollSec: R.pollSec }), { onMessage: () => false, onEpisode: onRouterEpisode, onStatus: () => {}, log, dataDir: ROUTER_DIR }); await routerConn.start(); log(`router: watching ${R.inbox}`); } // ---------------- plugins: inspect (sandboxed), AI draft, save ---------------- const PLUGIN_NAME = /^[a-z][a-z0-9-]{1,39}$/; const SHARED_PLUGINS = join(ROOT, 'plugins'); const BUILTIN_PLUGINS = join(ROOT, 'plugins', 'builtin'); const spacePlugins = (id) => join(spaceDir(id), 'plugins'); const INSPECT_JS = ` const { pathToFileURL } = await import('node:url'); const f = process.argv[1], want = process.argv[2]; try { const d = (await import(pathToFileURL(f).href)).default; const problems = []; if (!d || typeof d !== 'object') throw new Error('the file must "export default" an object'); if (d.name !== want) problems.push('name must be "' + want + '" (the file name), got "' + d.name + '"'); if (!d.description) problems.push('missing description'); for (const t of d.tools || []) { if (!t || !t.name || typeof t.run !== 'function') problems.push('every tool needs a name and a run() function'); else if (t.readOnly === undefined) problems.push('tool ' + t.name + ': set readOnly explicitly (false for anything with side effects)'); } for (const k of d.ticks || []) if (!k || typeof k.run !== 'function') problems.push('every tick needs a run() function'); for (const [k, v] of Object.entries(d.routes || {})) if (!/^(GET|POST|PUT|DELETE) \\/api\\/p\\//.test(k) || typeof v !== 'function') problems.push('route "' + k + '" must look like "GET /api/p//..." and map to a function'); console.log(JSON.stringify({ ok: problems.length === 0, problems, meta: { name: d.name, description: d.description || '', version: d.version || '', configSchema: d.configSchema || [], provides: { tools: (d.tools || []).map((t) => ({ name: t && t.name, readOnly: t && t.readOnly !== false })), ticks: (d.ticks || []).length, routes: Object.keys(d.routes || {}), connectors: (d.connectors || []).map((c) => c && c.type), domains: Object.keys(d.domains || {}), context: typeof d.context === 'function', setup: typeof d.setup === 'function' } } })); } catch (e) { console.log(JSON.stringify({ ok: false, problems: [String(e && e.message || e).slice(0, 300)] })); } process.exit(0);`; // import the plugin in a throwaway, permission-restricted node process and report its shape function inspectPlugin(file, name) { return new Promise((resolve) => { const args = ['--permission', `--allow-fs-read=${ROOT}`, `--allow-fs-read=${dirname(file)}`, '--input-type=module', '-e', INSPECT_JS, file, name]; const child = spawn(process.execPath, args, { cwd: ROOT, windowsHide: true, stdio: ['ignore', 'pipe', 'pipe'] }); let out = '', err = ''; const t = setTimeout(() => { child.kill(); resolve({ ok: false, problems: ['inspection timed out (top-level code must not block)'] }); }, 10000); child.stdout.on('data', (d) => (out += d)); child.stderr.on('data', (d) => (err += d)); child.on('exit', () => { clearTimeout(t); const line = out.trim().split('\n').pop(); try { resolve(JSON.parse(line)); } catch { resolve({ ok: false, problems: [(err || out || 'no output').trim().split('\n').slice(0, 3).join(' ').slice(0, 300)] }); } }); }); } async function listPlugins(id) { const cfg = loadCfg(id), out = []; for (const [scope, dir] of [['space', spacePlugins(id)], ['shared', SHARED_PLUGINS], ['builtin', BUILTIN_PLUGINS]]) { if (!existsSync(dir)) continue; for (const f of readdirSync(dir).filter((x) => x.endsWith('.js'))) { const name = f.slice(0, -3); if (out.some((p) => p.name === name)) continue; // space-local shadows shared const check = await inspectPlugin(join(dir, f), name); inspectCache.set(join(dir, f), { mtime: statSync(join(dir, f)).mtimeMs, result: check }); out.push({ name, scope, enabled: !!cfg.plugins?.[name] && cfg.plugins[name].enabled !== false, settings: await maskPluginSettings(id, name, cfg.plugins?.[name]), ...check }); } } return out; } async function draftPlugin(id, request, name) { const ai = settings.ai || {}; if (!(ai.base && ai.key && ai.model)) throw new Error('Set up the global AI provider first (Global settings)'); const c = loadCfg(id); const guide = readFileSync(join(ROOT, 'docs', 'PLUGINS.md'), 'utf8'); const example = readFileSync(join(BUILTIN_PLUGINS, 'keyword-alert.js'), 'utf8'); const space = { template: c.template, connectors: (c.connectors || []).map((x) => x.type), chats: (c.chats || []).map((x) => x.name).slice(0, 30) }; const raw = await aiChat(ai, [ { role: 'system', content: 'You write plugins for the Relate chat-assistant platform. Follow the guide exactly. Reply with ONE ```js code block containing the complete plugin file and nothing else.' }, { role: 'user', content: `# Guide\n${guide}\n\n# Example plugin\n\`\`\`js\n${example}\n\`\`\`\n\n# This space\n${JSON.stringify(space)}\n\n# Request\n${request}\n${name ? `\nThe plugin name (and file name) must be "${name}".` : '\nPick a short kebab-case name.'}` }, ], { maxTokens: 3500, timeout: 120000 }); const code = (raw.match(/```(?:js|javascript|mjs)?\s*\n([\s\S]*?)```/) || [null, raw])[1].trim() + '\n'; const nm = name || (code.match(/name:\s*['"]([a-z][a-z0-9-]{1,39})['"]/) || [])[1] || 'my-plugin'; const dir = join(HOME, '_scratch', 'drafts'); mkdirSync(dir, { recursive: true }); const f = join(dir, `${nm}.js`); writeFileSync(f, code); return { name: nm, code, ...(await inspectPlugin(f, nm)) }; } // ---------------- HTTP ---------------- const MIME = { '.html': 'text/html; charset=utf-8', '.js': 'text/javascript; charset=utf-8', '.css': 'text/css; charset=utf-8', '.json': 'application/json', '.svg': 'image/svg+xml', '.png': 'image/png' }; const json = (res, o, code = 200) => { res.writeHead(code, { 'Content-Type': 'application/json; charset=utf-8', 'Cache-Control': 'no-store' }); res.end(JSON.stringify(o)); }; function body(req, limit = 2e6) { return new Promise((ok) => { let b = ''; req.on('data', (c) => { b += c; if (b.length > limit) req.destroy(); }); req.on('end', () => { try { ok(b ? JSON.parse(b) : {}); } catch { ok({}); } }); }); } function serveStatic(res, p) { const rel = p === '/hub' || p === '/hub/' ? 'index.html' : p.slice('/hub/'.length); if (rel.includes('..')) return json(res, { error: 'bad path' }, 400); const f = join(ROOT, 'public', 'hub', rel); if (!existsSync(f) || !statSync(f).isFile()) return json(res, { error: 'not found' }, 404); res.writeHead(200, { 'Content-Type': MIME[extname(f)] || 'application/octet-stream', 'Cache-Control': 'no-cache' }); res.end(readFileSync(f)); } const server = http.createServer(async (req, res) => { const u = new URL(req.url, 'http://x'); const p = u.pathname; const M = req.method; try { if (p === '/' ) { res.writeHead(302, { Location: '/hub/' }); return res.end(); } if (p === '/hub' || (p.startsWith('/hub/') && !p.startsWith('/hub/api/'))) return serveStatic(res, p); if (!p.startsWith('/hub/api/')) return json(res, { error: 'not found' }, 404); const api = p.slice('/hub/api'.length); // --- unauthenticated --- if (api === '/session') return json(res, { authed: authed(req), setupNeeded: !settings.passwordHash, version: VERSION }); if (api === '/setup' && M === 'POST') { if (settings.passwordHash) return json(res, { ok: false, error: 'already set up' }, 409); const { password } = await body(req); if (!password || String(password).length < 8) return json(res, { ok: false, error: 'Use at least 8 characters' }, 400); settings.passwordHash = hashPw(password); writeJson(SETTINGS, settings); setCookie(res); return json(res, { ok: true }); } if (api === '/login' && M === 'POST') { const { password } = await body(req); if (!checkPw(password)) { await new Promise((ok) => setTimeout(ok, 400)); return json(res, { ok: false, error: 'Wrong password' }, 401); } setCookie(res); return json(res, { ok: true }); } if (api === '/logout' && M === 'POST') { res.setHeader('Set-Cookie', 'relate_hub=; Path=/; Max-Age=0'); return json(res, { ok: true }); } if (!authed(req)) return json(res, { error: 'unauthorized' }, 401); // --- meta --- if (api === '/meta') return json(res, { version: VERSION, templates: TEMPLATES, connectorTypes: await describeTypes() }); // --- spaces --- if (api === '/spaces' && M === 'GET') return json(res, await Promise.all(listIds().map(spaceSummary))); if (api === '/spaces' && M === 'POST') { const b = await body(req); const tpl = TEMPLATES.find((t) => t.id === b.template); if (!tpl) return json(res, { ok: false, error: 'Pick a template' }, 400); if (!String(b.name || '').trim()) return json(res, { ok: false, error: 'Name is required' }, 400); let id = b.id ? String(b.id) : slug(b.name); if (!validId(id)) return json(res, { ok: false, error: 'id: lowercase letters, digits and dashes' }, 400); if (existsSync(cfgFile(id))) { if (b.id) return json(res, { ok: false, error: `space "${id}" exists` }, 409); let n = 2; while (existsSync(cfgFile(`${id}-${n}`))) n++; id = `${id}-${n}`; } const c = { id, name: String(b.name).trim(), template: tpl.id, port: await freePort(), me: { name: b.me?.name || '' }, ...tpl.defaults, connectors: [], chats: [], ai: { base: '', key: '', model: '' }, auth: { password: '', secret: token(), apiToken: token() }, ingest: { token: token() }, plugins: Object.fromEntries((tpl.plugins || []).map((n) => [n, { enabled: true }])), autostart: false, createdAt: Date.now() }; mkdirSync(spaceDir(id), { recursive: true }); saveCfg(id, c); log(`space created: ${id} (${tpl.id}) on :${c.port}`); return json(res, { ok: true, id, space: await spaceSummary(id) }); } // --- plugins --- const pm = api.match(/^\/spaces\/([a-z0-9-]+)\/plugins(?:\/([a-z][a-z0-9-]{1,39}))?$/); if (pm && existsSync(cfgFile(pm[1]))) { const [, id, name] = pm; if (!name && M === 'GET') return json(res, await listPlugins(id)); if (name === 'draft' && M === 'POST') { const b = await body(req); if (!String(b.request || '').trim()) return json(res, { ok: false, problems: ['Describe what the plugin should do'] }, 400); if (b.name && !PLUGIN_NAME.test(b.name)) return json(res, { ok: false, problems: ['name: lowercase letters, digits, dashes'] }, 400); try { return json(res, await draftPlugin(id, String(b.request).slice(0, 4000), b.name)); } catch (e) { return json(res, { ok: false, problems: [e.message] }, 502); } } if (name && name !== 'draft') { const own = join(spacePlugins(id), `${name}.js`), shared = join(SHARED_PLUGINS, `${name}.js`), builtin = join(BUILTIN_PLUGINS, `${name}.js`); if (M === 'GET') { const f = [own, shared, builtin].find((x) => existsSync(x)); return f ? json(res, { name, scope: f === own ? 'space' : f === shared ? 'shared' : 'builtin', code: readFileSync(f, 'utf8') }) : json(res, { error: 'no such plugin' }, 404); } if (M === 'PUT') { const { code } = await body(req); if (!String(code || '').trim()) return json(res, { ok: false, problems: ['empty file'] }, 400); const tmpDir = join(HOME, '_scratch', 'check'); mkdirSync(tmpDir, { recursive: true }); const tmp = join(tmpDir, `${name}.js`); writeFileSync(tmp, code); const check = await inspectPlugin(tmp, name); if (!check.meta) return json(res, { ok: false, saved: false, problems: check.problems }, 400); // doesn't even load — refuse mkdirSync(spacePlugins(id), { recursive: true }); writeFileSync(own, code); return json(res, { ok: check.ok, saved: true, problems: check.problems, meta: check.meta, restartNeeded: ['running', 'starting'].includes(rec(id).status) }); } if (M === 'DELETE') { if (existsSync(own)) rmSync(own); const c = loadCfg(id); if (c.plugins?.[name]) { delete c.plugins[name]; saveCfg(id, c); } return json(res, { ok: true, restartNeeded: ['running', 'starting'].includes(rec(id).status) }); } } } const m = api.match(/^\/spaces\/([a-z0-9-]+)(\/[a-z-]+)?$/); if (m) { const [, id, sub = ''] = m; if (!existsSync(cfgFile(id))) return json(res, { error: 'no such space' }, 404); if (!sub && M === 'GET') return json(res, await maskCfg(loadCfg(id))); if (!sub && M === 'PUT') { const stored = loadCfg(id), b = await body(req); delete b.id; delete b.port; // hub-owned const next = await unmask(deepMerge(stored, b), stored); if (b.connectors) next.connectors = (await unmask({ connectors: b.connectors }, stored)).connectors; if (b.chats) next.chats = b.chats; const errors = await validateSpace(next); if (errors.length) return json(res, { ok: false, errors }, 400); saveCfg(id, next); const running = ['running', 'starting'].includes(rec(id).status); return json(res, { ok: true, restartNeeded: running }); } if (!sub && M === 'DELETE') { await stopSpace(id); if (u.searchParams.get('purge') === '1') rmSync(spaceDir(id), { recursive: true, force: true }); else renameSync(spaceDir(id), join(HOME, `_deleted-${id}-${Date.now()}`)); // recoverable by default procs.delete(id); log(`space deleted: ${id}`); return json(res, { ok: true }); } if (sub === '/start' && M === 'POST') { const errs = await validateSpace(loadCfg(id)); if (errs.length) return json(res, { ok: false, errors: errs }, 400); return json(res, { ok: true, status: await startSpace(id) }); } if (sub === '/stop' && M === 'POST') return json(res, { ok: true, status: await stopSpace(id) }); if (sub === '/restart' && M === 'POST') { await stopSpace(id, { remember: false }); return json(res, { ok: true, status: await startSpace(id) }); } if (sub === '/logs') { const n = Math.min(2000, Number(u.searchParams.get('tail')) || 200); return json(res, { lines: rec(id).lines.slice(-n) }); } if (sub === '/live') { if (rec(id).status !== 'running') return json(res, []); try { const r = await workerFetch(id, '/api/connectors'); return json(res, r.ok ? await r.json() : []); } catch { return json(res, []); } } } // --- connectors (test / chat picker, without saving) --- if (api === '/connectors/test' && M === 'POST') { const b = await body(req); return json(res, await withTempConnector(b, async (c) => (c.test ? await c.test() : { ok: true, info: 'no test available for this type' })).catch((e) => ({ ok: false, error: e.message }))); } if (api === '/connectors/chats' && M === 'POST') { const b = await body(req); const r = await withTempConnector(b, async (c) => (c.listChats ? await c.listChats() : [])).catch(() => []); return json(res, Array.isArray(r) ? r : []); } // --- global settings --- if (api === '/settings' && M === 'GET') return json(res, { ai: { ...settings.ai, key: settings.ai.key ? KEEP : '' }, episodeRouter: settings.episodeRouter, sandbox: !!settings.sandbox }); if (api === '/settings' && M === 'PUT') { const b = await body(req); if (b.ai) settings.ai = { ...settings.ai, ...b.ai, key: b.ai.key === KEEP ? settings.ai.key : (b.ai.key ?? settings.ai.key) }; if (b.episodeRouter) settings.episodeRouter = { ...settings.episodeRouter, ...b.episodeRouter }; if (typeof b.sandbox === 'boolean') settings.sandbox = b.sandbox; writeJson(SETTINGS, settings); if (b.episodeRouter) await startRouter().catch((e) => log('router start failed', e.message)); const running = listIds().filter((id) => rec(id).status === 'running'); return json(res, { ok: true, restartNeeded: !!b.ai && running.length > 0 }); } if (api === '/ai/test' && M === 'POST') { const b = await body(req); const spaceAi = b.space && validId(b.space) ? loadCfg(b.space)?.ai || {} : {}; const g = settings.ai || {}; const cfg = b.useGlobal ? g : { base: b.base || g.base, model: b.model || g.model, key: b.key === KEEP ? (spaceAi.key || g.key) : (b.key || g.key) }; if (!(cfg.base && cfg.key && cfg.model)) return json(res, { ok: false, error: 'base URL, key and model are all required' }); const t0 = Date.now(); try { const s = await aiChat(cfg, [{ role: 'user', content: 'Reply with the single word: ok' }], { maxTokens: 20, timeout: 45000 }); return json(res, { ok: true, ms: Date.now() - t0, sample: s.slice(0, 40) }); } catch (e) { return json(res, { ok: false, ms: Date.now() - t0, error: e.message.replace(cfg.key, '***') }); } } // --- recording review queue --- if (api === '/episodes/review') { const out = []; for (const f of readdirSync(EP_DIR).filter((x) => x.endsWith('.json'))) { const d = readJson(join(EP_DIR, f)); if (d?.routing?.status !== 'review') continue; out.push({ id: d.episode.id, title: d.episode.title, ts: d.episode.ts, excerpt: (d.episode.summary || d.episode.turns.slice(0, 4).map((t) => `${t.speaker}: ${t.text}`).join(' / ')).slice(0, 300), suggested: d.routing.suggested, error: d.routing.error || null }); } return json(res, out.sort((a, b) => b.ts - a.ts)); } const em = api.match(/^\/episodes\/([\w.-]+)\/route$/); if (em && M === 'POST') { const f = epFile(em[1]); const d = readJson(f); if (!d) return json(res, { error: 'no such recording' }, 404); const { spaces = [] } = await body(req); d.routing.pending = spaces.filter((s) => existsSync(cfgFile(s))); d.routing.status = spaces.length ? 'routed' : 'dismissed'; writeJson(f, d); for (const s of d.routing.pending) flushOutbox(s); return json(res, { ok: true, routed: d.routing.pending }); } return json(res, { error: 'not found' }, 404); } catch (e) { log('hub error', p, e.stack || e.message); if (!res.headersSent) json(res, { error: e.message }, 500); } }); async function main() { server.listen(PORT, HOST, () => log(`\n Relate hub -> http://${HOST}:${PORT}/hub/ (${listIds().length} spaces, home ${HOME})\n`)); for (const id of listIds()) if (loadCfg(id)?.autostart) startSpace(id).catch((e) => log(`autostart ${id} failed: ${e.message}`)); startRouter().catch((e) => log('router start failed', e.message)); } const shutdown = async () => { log('hub shutting down — stopping workers'); await Promise.all(listIds().map((id) => stopSpace(id, { remember: false }).catch(() => {}))); process.exit(0); }; process.on('SIGINT', shutdown); process.on('SIGTERM', shutdown); process.on('unhandledRejection', (e) => log('unhandledRejection', String(e?.stack || e))); main();