server.js
Rama main · réplica de lectura del repositorio original.
// Relate space worker: one space = one process. Connectors → store → Living State / todos / RAG / MCP /
// dashboard. The hub (hub.js) writes this space's config and starts it with RELATE_CONFIG=<runtime.json>.
import http from 'node:http';
import { readFileSync, writeFileSync, existsSync } from 'node:fs';
import { join, extname } from 'node:path';
import { CFG, ROOT, DATA_DIR, CONTEXT_TYPE, IS_BIZLIKE } from './lib/config.js';
import { createSources } from './lib/sources.js';
import { activatePlugins, pluginConnectorTypes } from './lib/plugins.js';
import * as S from './lib/store.js';
import { analyzeBatch, extractTodos, reconcileTodos, dedupeTodos, draftReply, askAssistant, askAssistantStream, askWithData, censusAnalyze, chatTitle, relationshipCheckup, regenerateArchive, regeneratePerspectives, observeLiving, groundLiving, groundBusiness, aiStats, aiBudget, aiDaily } from './lib/ai.js';
import { detectCensus, collectCensus } from './lib/census.js';
import * as chats from './lib/chats.js';
import { listNotes, addNote, deleteNote, notesText, extractRemember } from './lib/notes.js';
import { registerTool, executeTool, hasTool } from './lib/tools.js';
import { setExecutor, listApprovals, pendingApprovals, pendingCount, approveAction, rejectAction, addApproval } from './lib/approvals.js';
import { living, loadLiving, saveLiving, dismissLiving, activeMemory, dismissedList } from './lib/living.js';
import { sendPush } from './lib/notify.js';
import { vapidPublic, addSub, subCount } from './lib/webpush.js';
import * as auth from './lib/auth.js';
import { reminders, loadReminders, mergeExtracted, applyReconcile, collapseDuplicates, mergeAway, retireSensitive, markDone, clearDone, listReminders } from './lib/reminders.js';
import { windowDocsFrom } from './lib/windows.js';
import { Corpus } from './lib/rag.js';
import { initMcp, handleMcp } from './lib/mcp.js';
import { computeAnalytics, behaviorSnapshot } from './lib/analytics.js';
import { relTrends, trendText } from './lib/reltrends.js';
import { bizTrends, bizMood, bizMoodText } from './lib/biztrends.js';
import { computeGame } from './lib/game.js';
const PUB = join(ROOT, 'public');
const clients = new Set();
let sources = null; // connector runtime (lib/sources.js)
let plugins = { route: () => null, contextText: () => '', names: [] }; // lib/plugins.js runtime
let corpus = new Corpus();
const readData = (f, dflt = null) => { try { return JSON.parse(readFileSync(join(DATA_DIR, f), 'utf8')); } catch { return dflt; } };
const writeData = (f, v) => { try { writeFileSync(join(DATA_DIR, f), JSON.stringify(v)); } catch {} };
let gameState = readData('game.json', { questDone: {} });
let narrative = readData('narrative.json');
let regen = readData('regen.json');
let checkup = readData('checkup.json');
// the space's chat allowlist (the hub writes each entry's store session id) + a name map for labeling
const SESSIONS = new Map(); // sid -> {name, kind}
const NAMES = {};
for (const c of CFG.chats || []) { const sid = c.session || c.chatId; SESSIONS.set(sid, { name: c.name || sid, kind: c.kind || 'dm' }); if (c.name) NAMES[sid] = c.name; }
let groundTimer = null;
function bumpGroundSoon() { if (groundTimer) return; groundTimer = setTimeout(() => { groundTimer = null; groundBizTick(); }, 90000); }
// a "deep" question gets more retrieved windows + a longer memory slice
const DEEP_RE = /深入|详细|彻底|全面|系统|完整|仔细|好好|认真|所有|每一|每个|逐条|统计|挨个|分析一下|深挖|deep|thorough|detailed|comprehensive|in depth/i;
const isDeep = (q) => DEEP_RE.test(String(q || '')) || String(q || '').length > 40;
// Assemble the grounding context the advisor/checkup reasons from: Living State + real stats.
function assembleCtx() {
let snapshot = null;
// behavioral snapshot uses the relationship baseline — skip it for multi-chat spaces
if (!IS_BIZLIKE) { try { snapshot = behaviorSnapshot(computeAnalytics(S.store.messages), CFG.target, S.store.messages); } catch {} }
// the actual latest messages, verbatim, so "right now" questions see the real exchange (not just a summary)
const recent = S.store.messages.slice(-60)
.filter((m) => String(m.content || '').trim())
.slice(-30)
.map((m) => ({
who: m.isSend ? CFG.me.name : (m.senderName || CFG.target?.name || '对方'),
text: (m.kind && m.sessionName ? `[${m.sessionName}] ` : '') + String(m.content).slice(0, 200),
ts: m.ts,
}));
const fin = finishedItems();
const done = fin.length ? '【最近已完成/已处理（别再当作待办、承诺、未完成或重复提醒）】\n' + fin.map((t) => '· ' + t).join('\n') : null;
let trends = null, contacts = null;
if (!IS_BIZLIKE) { try { trends = trendText(relTrends(S.store.messages, Date.now())) || null; } catch {} }
else { try { contacts = bizMoodText(bizMood(S.store.messages, SESSIONS, Date.now())) || null; } catch {} }
return { memory: living.memory, narrative: (narrative?.narrative || living.narrative), observation: living.observation, snapshot, recent, self: CFG.me.about || null, metrics: plugins.contextText() || null, notes: notesText() || null, done, trends, contacts };
}
// recently finished work — ticked-off Living-State lines + completed todos — so the AI treats them as done
function finishedItems() {
const dismissed = dismissedList().map((d) => d.text);
let doneTodos = [];
try { doneTodos = (listReminders().done || []).slice(0, 12).map((r) => r.text); } catch {}
return [...new Set([...dismissed, ...doneTodos])].filter(Boolean).slice(0, 20);
}
let voiceChatId = null; // persistent chat for the voice endpoint (continuous conversation)
// flatten markdown to plain text so a voice assistant reads it naturally (no "asterisk asterisk", pipes, headers…)
function stripMarkdown(s) {
return String(s || '')
.replace(/```[\s\S]*?```/g, '').replace(/`([^`]*)`/g, '$1')
.replace(/^\s*#{1,6}\s*/gm, '').replace(/\*\*([^*]+)\*\*/g, '$1').replace(/\*([^*]+)\*/g, '$1')
.replace(/^\s*[-*+]\s+/gm, '').replace(/^\s*\d+\.\s+/gm, '')
.replace(/^\s*[-:|\s]{3,}\s*$/gm, '').replace(/\|/g, ' ')
.replace(/\[(\d+)\]/g, '').replace(/\[([^\]]+)\]\([^)]+\)/g, '$1')
.replace(/[ \t]{2,}/g, ' ').replace(/\n{3,}/g, '\n\n').trim();
}
function log(...a) { console.log(new Date().toISOString().slice(11, 19), ...a); }
function broadcast(event, data) {
const p = `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`;
for (const res of clients) { try { res.write(p); } catch {} }
}
function buildCorpus() {
const docs = windowDocsFrom(S.store.messages);
corpus = new Corpus();
for (const d of docs) corpus.add(d);
sources?.reindexEpisodes(corpus);
corpus.finalize();
log(`RAG corpus ${corpus.size} conversation windows · ${S.store.messages.length} total messages`);
}
// ---- AI loops (low frequency, hourly-capped) ----
let busy = { an: false, sum: false, rem: false };
async function analysisTick() {
if (busy.an || !CFG.ai.enabled || noViewer()) return; busy.an = true; // pause when nobody's watching (cost)
try {
const batch = S.pendingAnalysis(CFG.ai.batchSize);
if (batch.length) {
const r = await analyzeBatch(batch);
let k = 0; r.forEach((x, i) => { if (x) { batch[i].analysis = x; k++; } });
if (k) broadcast('stats', S.snapshot());
}
} catch (e) { log('analysis err', e.message); } finally { busy.an = false; }
}
// one-shot burst: analyze recent unanalyzed messages fast (fills the sentiment trend now instead of
// waiting for the slow loop). Newest-first, bounded by weeks + a batch cap.
let backfilling = false;
async function backfillAnalysis(weeks = 8, maxBatches = 600) {
if (backfilling || !CFG.ai.enabled) return { ok: false };
backfilling = true; let done = 0;
const cutoff = Date.now() - weeks * 7 * 86400000;
try {
for (let n = 0; n < maxBatches; n++) {
const batch = S.pendingAnalysis(CFG.ai.batchSize).filter((m) => m.ts >= cutoff);
if (!batch.length) break;
const r = await analyzeBatch(batch);
let k = 0; r.forEach((x, i) => { if (x) { batch[i].analysis = x; k++; } });
if (!k) break; // rate-limited / failing — stop rather than spin
done += k; if (n % 8 === 0) broadcast('stats', S.snapshot());
}
} finally { backfilling = false; }
broadcast('stats', S.snapshot());
log(`backfill analysis: +${done} msgs (${weeks}w window)`);
return { ok: true, analyzed: done };
}
// pause the expensive AI loops when nobody is viewing the dashboard (huge idle-cost saver)
const noViewer = () => CFG.ai.pauseWhenNoViewer && clients.size === 0;
let sumMark = -1, remMark = -1;
async function summaryTick() {
if (busy.sum || !CFG.ai.enabled) return;
// seed the Living State once on boot even with no viewer (so the board/observation isn't empty);
// after that, pause when nobody's watching to save cost. Multi-chat spaces use groundBizTick instead.
const seedOnce = !living.at && !IS_BIZLIKE;
if (!seedOnce && noViewer()) return;
if (S.store.messages.length === sumMark) return; // idle-skip: nothing new
sumMark = S.store.messages.length; busy.sum = true;
try {
// incremental: fold recent messages into the Living State memory + produce 实时观察.
// multi-chat spaces need a wider recent window than a 1:1 relationship.
const tail = S.store.messages.slice(IS_BIZLIKE ? -60 : -20);
const r = await observeLiving(living.memory, tail);
if (r) {
living.memory = r.memory; living.at = Date.now();
if (r.observation) { living.observation = r.observation; saveLiving(); S.store.ai.insight = r.observation; S.store.ai.insightAt = Date.now(); broadcast('insight', { insight: r.observation, at: S.store.ai.insightAt }); }
else saveLiving();
} else log('observe: null (parse/empty/rate-limited)');
} catch (e) { log('observe err', e.message); } finally { busy.sum = false; }
}
// adaptive 实时观察 cadence: calm ~5min, warming ~2min, heat/crisis ~30s (near real-time)
let summaryTimer = null, summaryFireAt = 0;
function recentHeat() {
const now = Date.now(); let h = 0, care = false;
for (const m of S.store.messages) if (m.ts >= now - 8 * 60000) { h += (m._tag?.conflict || 0); if (m._tag?.care) care = true; }
return { h, care };
}
function summaryDelay() {
const { h, care } = recentHeat();
if (care || h >= 15) return CFG.ai.summaryHeatMs || 30000; // 激烈/危机 → 近实时
if (h >= 5) return 120000; // 升温 → 2分钟
return CFG.ai.summaryCalmMs || 300000; // 平静 → 5分钟
}
function armSummary(delay) {
if (summaryTimer) clearTimeout(summaryTimer);
summaryFireAt = Date.now() + delay;
summaryTimer = setTimeout(async () => { await summaryTick(); armSummary(summaryDelay()); }, delay);
}
// event-driven: a conflict/care message just arrived → pull the next observation in soon
function bumpSummaryOnHeat(tag) {
if (!tag || !(tag.conflict > 0 || tag.care)) return;
const soon = Date.now() + (CFG.ai.summaryHeatMs || 30000);
if (soon < summaryFireAt) armSummary(CFG.ai.summaryHeatMs || 30000);
}
async function reminderTick() {
if (busy.rem || !CFG.ai.enabled) return;
const seedOnce = remMark === -1 && !reminders.size; // seed todos once on boot even w/o viewer
if (!seedOnce && noViewer()) return;
if (S.store.messages.length === remMark) return; // idle-skip: nothing new
remMark = S.store.messages.length; busy.rem = true;
try {
const label = CFG.target.name || CFG.name || '';
const items = await extractTodos(S.store.messages, label, !IS_BIZLIKE);
const added = items.length ? mergeExtracted(items, Date.now(), 'main', label) : 0;
if (added) log(`+${added} todos`);
broadcast('reminders', listReminders());
} catch (e) { log('reminder err', e.message); } finally { busy.rem = false; }
}
// daily re-grounding: rebuild Living State memory from raw + refresh 近期动态 (1 AI call)
let busyNar = false;
async function narrativeTick() {
if (busyNar || !CFG.ai.enabled) return; busyNar = true;
log('ground/narrative run');
try {
const recent = S.store.messages.filter((m) => m.ts >= Date.now() - 7 * 86400000);
const r = await groundLiving(recent.length >= 5 ? recent : S.store.messages);
if (!r) { log('ground: null (parse/empty)'); return; }
living.memory = r.memory; living.narrative = r.narrative; living.groundedAt = Date.now(); saveLiving();
narrative = r.narrative;
writeData('narrative.json', r.narrative);
broadcast('narrative', r.narrative);
log('living re-grounded + narrative refreshed');
} catch (e) { log('ground err', e.message); } finally { busyNar = false; }
}
// FULL regeneration of 档案 + 视角 (2 AI calls). Runs on boot if stale + daily at 20:00.
let busyRegen = false;
async function regenTick() {
if (busyRegen || !CFG.ai.enabled) return; busyRegen = true;
log('regenTick: regenerating 档案 + 视角');
try {
const recent = S.store.messages.filter((m) => m.ts >= Date.now() - 30 * 86400000);
const [archive, perspectives] = await Promise.all([regenerateArchive(recent), regeneratePerspectives(recent)]);
regen = { archive: archive || regen?.archive || null, perspectives: perspectives || regen?.perspectives || null, at: Date.now() };
writeData('regen.json', regen);
broadcast('regen', regen);
log(`regen done: ${archive?.episodes?.length || 0} episodes, perspectives ${perspectives ? 'ok' : 'fail'}`);
} catch (e) { log('regen err', e.message); } finally { busyRegen = false; }
}
let busyCheckup = false;
async function checkupTick() {
if (busyCheckup || !CFG.ai.enabled) return; busyCheckup = true;
log('checkupTick: generating 关系体检');
try {
// sample broadly: recent windows + conflict/care-heavy windows for evidence
const windows = corpus ? corpus.search('吵架 冲突 生气 难过 开心 想你 道歉 在乎 期待 未来', 12) : [];
const r = await relationshipCheckup(assembleCtx(), windows);
if (r && r.report) {
checkup = r;
writeData('checkup.json', checkup);
broadcast('checkup', checkup);
log('checkup done');
}
} catch (e) { log('checkup err', e.message); } finally { busyCheckup = false; }
}
function msUntil(hour) {
const now = new Date(); const t = new Date(now); t.setHours(hour, 0, 0, 0);
if (t <= now) t.setDate(t.getDate() + 1);
return t - now;
}
// reconcile open todos against your subsequent replies — auto-stash addressed/stale (1 call)
let busyRec = false, recMark = -1;
async function reconcileTick() {
if (busyRec || !CFG.ai.enabled) return;
const collapsed = collapseDuplicates(Date.now()); // local, no AI — drain reworded duplicates
if (collapsed) { log(`collapsed ${collapsed} duplicate todos`); broadcast('reminders', listReminders()); }
const seedOnce = recMark === -1; // run once on boot even with no viewer (clean the seed)
if (!seedOnce && noViewer()) return; // otherwise pause when nobody's watching
// AI dedupe: merge semantically-duplicate todos the bigram heuristic misses (keep highest-scored per group)
try {
const openNow = listReminders().open.slice(0, 40);
if (openNow.length >= 3) {
const groups = await dedupeTodos(openNow.map((r, i) => ({ i, text: r.text })));
let dm = 0;
for (const g of groups) {
const items = g.map((idx) => openNow[idx]).filter(Boolean);
if (items.length < 2) continue;
items.sort((a, b) => (b.score || 0) - (a.score || 0)); // keep the best-scored, stale the rest
dm += mergeAway(items.slice(1).map((r) => r.id), Date.now());
}
if (dm) { log(`AI-merged ${dm} duplicate todos`); broadcast('reminders', listReminders()); }
}
} catch (e) { log('dedupe err', e.message); }
if (S.store.messages.length === recMark) return; // idle-skip the AI pass
// reconcile the RESOLVABLE ones first: reply/discuss (answered?) > promise (done?) > sensitive
const pri = { reply: 0, discuss: 1, promise: 2, sensitive: 3 };
const open = listReminders().open
.sort((a, b) => (pri[a.kind] - pri[b.kind]) || ((b.lastSeen || 0) - (a.lastSeen || 0))).slice(0, 20);
if (open.length < 2) return;
recMark = S.store.messages.length; busyRec = true;
try {
const todos = open.map((r, i) => ({ i, kind: r.kind, text: r.text, id: r.id }));
const idByIndex = Object.fromEntries(todos.map((t) => [t.i, t.id]));
// RAG: pull corpus-wide evidence for each todo so completion isn't limited to the recent tail
const evidence = {};
if (corpus) for (const t of todos) { const h = corpus.search(t.text, 4); if (h.length) evidence[t.i] = h.map((x) => String(x.text).replace(/\s+/g, ' ').slice(0, 140)).join(' ／ '); }
const verdicts = await reconcileTodos(todos, S.store.messages, evidence);
log(`reconcile verdict: addressed=${JSON.stringify(verdicts.addressed)} stale=${JSON.stringify(verdicts.stale)} (of ${todos.length})`);
const changed = applyReconcile(verdicts, idByIndex, Date.now());
if (changed) { log(`reconciled ${changed} todos (addressed/stale)`); broadcast('reminders', listReminders()); }
} catch (e) { log('reconcile err', e.message); } finally { busyRec = false; }
}
// multi-chat grounding: build the full per-chat state (feeds the 概览 board). boot + daily.
let busyGround = false;
async function groundBizTick() {
if (busyGround || !CFG.ai.enabled) return; busyGround = true;
try {
const r = await groundBusiness(S.store.messages, finishedItems());
if (r && (r.memory || r.narrative)) {
if (r.memory) { living.memory = r.memory; living.at = Date.now(); }
if (r.narrative) { living.narrative = r.narrative; narrative = { narrative: r.narrative, at: Date.now() }; broadcast('narrative', narrative); }
const obs = { headline: String(r.memory?.focus || '当前状态').slice(0, 20), summary: r.narrative || '', mood: '', activeTopics: [], alerts: (r.memory?.watchNow || []).slice(0, 3) };
living.observation = obs; S.store.ai.insight = obs; S.store.ai.insightAt = Date.now();
saveLiving();
broadcast('insight', { insight: obs, at: S.store.ai.insightAt });
log('ground: multi-chat state grounded');
}
} catch (e) { log('ground err', e.message); } finally { busyGround = false; }
}
// ---- HTTP ----
const MIME = { '.html': 'text/html; charset=utf-8', '.js': 'text/javascript; charset=utf-8', '.css': 'text/css', '.json': 'application/json', '.png': 'image/png', '.svg': 'image/svg+xml' };
function serveFile(res, full) {
const ext = extname(full);
// live dashboard: HTML/JS/CSS must always revalidate so UI updates aren't stuck on stale cache.
const noCache = ext === '.html' || ext === '.js' || ext === '.css';
res.writeHead(200, {
'Content-Type': MIME[ext] || 'application/octet-stream',
'Cache-Control': noCache ? 'no-cache, must-revalidate' : 'public, max-age=86400',
});
res.end(readFileSync(full));
}
function json(res, obj, code = 200) { res.writeHead(code, { 'Content-Type': 'application/json; charset=utf-8' }); res.end(JSON.stringify(obj)); }
function body(req) { return new Promise((resolve) => { let b = ''; req.on('data', (c) => (b += c)); req.on('end', () => { try { resolve(JSON.parse(b || '{}')); } catch { resolve({}); } }); }); }
// a request straight from this computer: loopback socket, loopback Host, and no proxy/tunnel headers
function isLocalRequest(req) {
const ip = req.socket.remoteAddress || '';
const host = String(req.headers.host || '').replace(/:\d+$/, '');
const proxied = ['x-forwarded-for', 'x-forwarded-host', 'x-real-ip', 'forwarded', 'cf-connecting-ip', 'tailscale-user-login'].some((h) => req.headers[h]);
return /^(127\.|::1$|::ffff:127\.)/.test(ip) && /^(127\.0\.0\.1|localhost|\[::1\])$/.test(host) && !proxied;
}
const NEEDS_PASSWORD_HTML = `<!doctype html><html lang="zh-CN"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1"><title>Relate</title>
<style>body{margin:0;min-height:100vh;display:flex;align-items:center;justify-content:center;background:#faf8f5;font:15px/1.7 -apple-system,"PingFang SC",sans-serif;color:#2d2a26}div{max-width:420px;padding:24px}</style></head>
<body><div><h2>先设置看板密码</h2><p>这个空间还没有密码，所以只能在运行 Relate 的电脑上打开。请在控制台（hub）→ 空间 → <b>访问</b> 里设置看板密码，保存并重启空间后再从手机访问。</p>
<p style="color:#7a746c">Set a dashboard password first: this space has none, so it only opens on the computer running Relate. Hub → Space → <b>Access</b> → set a password, save and restart the space.</p></div></body></html>`;
// public, non-secret description of this space for the dashboard (names, template, enabled plugins)
function spaceInfo() {
const t = CFG.target || {};
return {
id: CFG.id || '', name: CFG.name || '', template: CONTEXT_TYPE,
me: { name: CFG.me.name && CFG.me.name !== 'me' ? CFG.me.name : '' },
target: { name: t.name || '', pronoun: t.pronoun || '', togetherDate: t.togetherDate || '', firstContactDate: t.firstContactDate || '' },
org: { name: CFG.org?.name || '' },
plugins: plugins.names,
};
}
const server = http.createServer(async (req, res) => {
const u = new URL(req.url, 'http://x');
const p = u.pathname;
// ---- ingest: bridges push canonical messages/episodes with the connector's bearer token ----
if (req.method === 'POST' && (p === '/api/ingest' || p.startsWith('/api/ingest/'))) {
const want = decodeURIComponent(p.slice('/api/ingest/'.length) || '');
const c = want ? sources?.get(want) : sources?.firstOfType('ingest');
if (!c || c.type !== 'ingest') return json(res, { ok: false, error: 'no ingest connector' + (want ? ` "${want}"` : '') }, 404);
const bearer = String(req.headers.authorization || '').replace(/^Bearer\s+/i, '');
if (!c.checkToken(bearer)) return json(res, { ok: false, error: 'bad or missing ingest token' }, 401);
const r = c.accept(await body(req));
if (r.added) broadcast('stats', S.snapshot());
return json(res, r, r.ok ? 200 : 400);
}
// ---- no password = this computer only: a tunnel, proxy or LAN visitor never sees an unprotected space ----
if (!auth.authEnabled() && !isLocalRequest(req)) {
res.writeHead(403, { 'Content-Type': 'text/html; charset=utf-8' });
return res.end(NEEDS_PASSWORD_HTML);
}
// ---- auth gate ----
if (p === '/login') {
if (req.method === 'POST') {
let pw = ''; try { pw = JSON.parse(await auth.readBody(req)).password; } catch {}
if (auth.checkPassword(pw)) { auth.setAuthCookie(res); return json(res, { ok: true }); }
return json(res, { ok: false }, 401);
}
res.writeHead(200, { 'Content-Type': 'text/html; charset=utf-8' }); return res.end(auth.LOGIN_HTML);
}
if (auth.authEnabled() && !auth.isPublicPath(p) && !auth.isAuthed(req)) {
if (p.startsWith('/api/') || p === '/stream' || p === '/mcp' || req.method !== 'GET') return json(res, { error: 'unauthorized' }, 401);
res.writeHead(200, { 'Content-Type': 'text/html; charset=utf-8' }); return res.end(auth.LOGIN_HTML);
}
// MCP service: every chat + the board, read-only, for Claude Code / Desktop / scripts (Bearer apiToken)
if (p === '/mcp') { try { await handleMcp(req, res, req.method === 'POST' ? await body(req) : undefined); } catch (e) { log('mcp err', e.message); if (!res.headersSent) json(res, { error: 'mcp failure' }, 500); } return; }
if (p === '/stream') {
res.writeHead(200, { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache', 'Connection': 'keep-alive' });
res.write('retry: 3000\n\n');
res.write(`event: stats\ndata: ${JSON.stringify(S.snapshot())}\n\n`);
res.write(`event: reminders\ndata: ${JSON.stringify(listReminders())}\n\n`);
const wasEmpty = clients.size === 0;
clients.add(res);
// first viewer arrived → refresh soon so the dashboard isn't stale when you open it
if (wasEmpty && CFG.ai.enabled) { armSummary(2000); setTimeout(reminderTick, 5000); setTimeout(reconcileTick, 9000); }
const ping = setInterval(() => { try { res.write(': ping\n\n'); } catch {} }, 20000);
req.on('close', () => { clients.delete(res); clearInterval(ping); });
return;
}
{ const h = plugins.route(req.method, p); if (h) { try { return await h(req, res); } catch (e) { log('plugin route err', e.message); return json(res, { error: e.message }, 500); } } }
if (p === '/api/space') return json(res, spaceInfo());
if (p === '/api/state') return json(res, S.snapshot());
// ---- connectors, send, episodes ----
if (p === '/api/connectors') return json(res, sources ? sources.describe() : []);
if (sources && p.startsWith('/api/connectors/')) {
const [, , , cid, action] = p.split('/');
const c = sources.get(decodeURIComponent(cid));
if (!c) return json(res, { ok: false, error: 'unknown connector' }, 404);
if (action === 'test' && req.method === 'POST') return json(res, c.test ? await c.test().catch((e) => ({ ok: false, error: e.message })) : { ok: true, info: 'no test available' });
if (action === 'chats') return json(res, c.listChats ? await c.listChats().catch(() => []) : []);
return json(res, { error: 'not found' }, 404);
}
if (p === '/api/send' && req.method === 'POST') {
if (!sources) return json(res, { ok: false, error: 'connectors not started' }, 503);
const b = await body(req);
return json(res, await sources.send({ chatId: b.chatId, text: b.text, connector: b.connector }).catch((e) => ({ ok: false, error: e.message })));
}
if (p === '/api/episodes') return json(res, sources ? sources.listEpisodes(Number(u.searchParams.get('limit')) || 50) : []);
if (p === '/api/analytics') return json(res, { ...computeAnalytics(S.store.messages), target: spaceInfo().target });
if (p === '/api/trends') return json(res, !IS_BIZLIKE ? relTrends(S.store.messages, Date.now()) : {});
if (p === '/api/contacts') return json(res, IS_BIZLIKE ? { contacts: bizTrends(S.store.messages, SESSIONS, Date.now()) } : { contacts: [] });
if (p === '/api/mood') return json(res, IS_BIZLIKE ? bizMood(S.store.messages, SESSIONS, Date.now()) : {});
if (p === '/api/analyze/backfill' && req.method === 'POST') { const b = await body(req); backfillAnalysis(b.weeks || 8).catch((e) => log('backfill err', e.message)); return json(res, { ok: true, note: '已开始补分析（后台进行，情绪趋势会陆续填充）' }); }
if (p === '/api/game') return json(res, computeGame(computeAnalytics(S.store.messages), S.store.messages, gameState));
if (p === '/api/narrative') return json(res, narrative || living.narrative || {});
if (p === '/api/living') return json(res, { memory: activeMemory(), at: living.at, groundedAt: living.groundedAt });
if (p === '/api/living/dismiss' && req.method === 'POST') { const b = await body(req); const ok = dismissLiving(b.text); if (ok) broadcast('insight', { insight: living.observation, at: S.store.ai.insightAt }); return json(res, { ok }); }
if (p === '/api/notify/test') { const r = await sendPush('Relate', '✅ 测试通知：推送已连通', { click: CFG.notify?.dashboardUrl }); return json(res, r); }
if (p === '/api/push/vapid') return json(res, { key: vapidPublic(), subs: subCount() });
if (p === '/api/push/subscribe' && req.method === 'POST') { const b = await body(req); const n = addSub(b.subscription || b); log(`push subscription added (${n} total)`); return json(res, { ok: true, subs: n }); }
if (p === '/api/regen') return json(res, regen || {});
if (p === '/api/regen/run' && req.method === 'POST') { regenTick(); return json(res, { ok: true, note: '已触发重新生成' }); }
if (p === '/api/notes' && req.method === 'GET') return json(res, { notes: listNotes() });
if (p === '/api/notes' && req.method === 'POST') { const b = await body(req); const n = addNote(b.text); return json(res, n ? { ok: true, note: n } : { ok: false }, n ? 200 : 400); }
if (p.startsWith('/api/notes/') && req.method === 'DELETE') { const ok = deleteNote(decodeURIComponent(p.slice('/api/notes/'.length))); return json(res, { ok }); }
// approval queue: proposed side-effectful actions await your ✓ here before executing
if (p === '/api/approvals') return json(res, { pending: pendingApprovals(), recent: listApprovals(30) });
if (p.startsWith('/api/approvals/') && p.endsWith('/approve') && req.method === 'POST') {
const r = await approveAction(decodeURIComponent(p.slice('/api/approvals/'.length, -'/approve'.length)));
broadcast('approvals', { pending: pendingCount() }); return json(res, r);
}
if (p.startsWith('/api/approvals/') && p.endsWith('/reject') && req.method === 'POST') {
const r = rejectAction(decodeURIComponent(p.slice('/api/approvals/'.length, -'/reject'.length)));
broadcast('approvals', { pending: pendingCount() }); return json(res, r);
}
if (p === '/api/checkup') return json(res, checkup || { report: '', at: null });
if (p === '/api/checkup/run' && req.method === 'POST') { checkupTick(); return json(res, { ok: true, note: '正在生成关系体检…' }); }
if (p.startsWith('/api/game/quest/') && p.endsWith('/toggle') && req.method === 'POST') {
const id = decodeURIComponent(p.slice('/api/game/quest/'.length, -'/toggle'.length));
const tk = new Date().toISOString().slice(0, 10);
gameState.questDone = gameState.questDone || {};
gameState.questDone[tk] = gameState.questDone[tk] || {};
gameState.questDone[tk][id] = !gameState.questDone[tk][id];
writeData('game.json', gameState);
return json(res, { ok: true });
}
if (p === '/api/reminders') return json(res, listReminders());
if (p === '/api/ai') return json(res, { ...aiStats, ...aiBudget(), daily: aiDaily(), enabled: CFG.ai.enabled, model: CFG.ai.model, corpus: corpus.size });
// ---- multi-chat: list / open / create / delete persistent 问它 conversations ----
if (p === '/api/chats' && req.method === 'GET') return json(res, { chats: chats.listChats() });
if (p === '/api/chats' && req.method === 'POST') { const c = chats.createChat(); return json(res, { id: c.id, title: c.title, messages: [] }); }
if (p.startsWith('/api/chats/') && req.method === 'GET') {
const id = decodeURIComponent(p.slice('/api/chats/'.length));
const c = chats.getChat(id); if (!c) return json(res, { error: 'not found' }, 404);
return json(res, { id: c.id, title: c.title, messages: c.messages });
}
if (p.startsWith('/api/chats/') && req.method === 'DELETE') {
chats.deleteChat(decodeURIComponent(p.slice('/api/chats/'.length))); return json(res, { ok: true });
}
// append the finished exchange to a chat + auto-title on first exchange
async function persistExchange(cid, q, answer, hits, cited) {
if (!cid) return;
chats.appendMessage(cid, { role: 'user', content: q });
chats.appendMessage(cid, { role: 'assistant', content: answer, hits: hits || null, cited: cited || null });
const c = chats.getChat(cid);
// title in the BACKGROUND — never block the reply on it
if (c && !c.title) { chatTitle(q, answer).then((t) => { if (t) chats.setTitle(cid, t); }).catch(() => {}); }
}
// data mode = the assistant may call tools to fetch numbers (e.g. the sql-readonly plugin's run_sql)
const dataMode = () => IS_BIZLIKE && hasTool('run_sql');
if (p === '/api/ask') {
let q = (u.searchParams.get('q') || '').trim(), history = [];
let chatId = null;
if (req.method === 'POST') { const b = await body(req); q = String(b.q || '').trim(); chatId = b.chatId || null; }
if (!q) return json(res, { error: 'empty' }, 400);
// resolve/create chat + server-authoritative history (consistent with the streaming endpoint)
const cid = (chatId && chats.getChat(chatId)) ? chatId : chats.createChat().id;
// "记住 X" → save a background fact instead of asking
{ const fact = extractRemember(q); if (fact) { const n = addNote(fact); const ans = n ? `✅ 记住了：${fact}\n\n（以后回答都会考虑这条。可在设置里查看/删除。）` : '这条太短，没记下来。'; await persistExchange(cid, q, ans, [], []); return json(res, { answer: ans, cited: [], hits: [], chatId: cid, remembered: !!n }); } }
history = chats.historyFor(cid);
// non-streaming fallback also handles census (returns the final report, no live progress)
const census0 = detectCensus(q);
if (census0) {
try { const { report } = await censusAnalyze(census0, collectCensus(S.store.messages, census0), assembleCtx(), {}); await persistExchange(cid, q, report, [], []); return json(res, { answer: report, cited: [], hits: [], chatId: cid }); }
catch (e) { return json(res, { answer: '(分析失败) ' + e.message, hits: [] }, 500); }
}
// retrieve on the last couple of turns + current q so follow-ups ("那怎么办") still find context
const retrievalQ = [...history.filter((h) => h.role === 'user').slice(-2).map((h) => h.content), q].join(' ');
const deep = isDeep(q);
const hits = corpus.search(retrievalQ, deep ? 16 : 12);
const hitMeta = hits.map((h, i) => ({ i, date: h.meta.date, dateEnd: h.meta.dateEnd, text: h.text.slice(0, 240), score: h.score }));
try {
const { answer, cited } = dataMode()
? await askWithData(q, hits, assembleCtx(), history, { deep })
: await askAssistant(q, hits, assembleCtx(), history, deep);
await persistExchange(cid, q, answer, hitMeta, cited);
return json(res, { answer, cited, hits: hitMeta, chatId: cid });
} catch (e) { return json(res, { answer: '(AI 调用失败) ' + e.message, hits: hitMeta }, 500); }
}
// voice endpoint (e.g. an iOS Shortcut): one persistent "voice" chat, concise no-markdown answers
// for text-to-speech. Auth via cookie OR Bearer token. Returns { answer } as plain spoken text.
if (p === '/api/voice' && (req.method === 'POST' || req.method === 'GET')) {
let raw = ''; if (req.method === 'POST') { try { raw = await auth.readBody(req); } catch {} }
let q = '', reset = false;
try { const j = JSON.parse(raw); if (j && j.q != null) q = String(j.q); if (j && j.reset) reset = true; } catch {} // JSON body {q}
if (!q && raw && !raw.trim().startsWith('{')) q = raw; // plain-text body
if (!q) q = u.searchParams.get('q') || ''; // ?q= (also works for GET)
q = q.trim();
if (reset) { voiceChatId = null; if (!q) return json(res, { answer: 'OK, starting fresh.' }); }
if (!q) return json(res, { answer: "I didn't catch a question — say it again." }); // friendly, so the voice assistant speaks something
if (!voiceChatId || !chats.getChat(voiceChatId)) voiceChatId = chats.createChat('🎙️ 语音').id;
const cid = voiceChatId;
{ const fact = extractRemember(q); if (fact) { const n = addNote(fact); const ans = n ? `Got it — I'll remember that: ${fact}` : 'That was too short to save.'; await persistExchange(cid, q, ans, [], []); return json(res, { answer: ans, chatId: cid }); } }
const history = chats.historyFor(cid);
const retrievalQ = [...history.filter((h) => h.role === 'user').slice(-2).map((h) => h.content), q].join(' ');
const hits = corpus.search(retrievalQ, 12);
try {
const vq = q + '\n\n[Answer in spoken English only — brief, no symbols. This is read aloud.]';
const { answer } = dataMode()
? await askWithData(vq, hits, assembleCtx(), history, { voice: true, fast: true })
: await askAssistant(vq, hits, assembleCtx(), history, false, true);
const spoken = stripMarkdown(answer);
await persistExchange(cid, q, spoken, [], []);
return json(res, { answer: spoken, chatId: cid });
} catch (e) { return json(res, { answer: '出错了：' + e.message }, 500); }
}
// streaming ask: NDJSON — {type:'meta'|'token'|'progress'|'done'|'error'}. Headers flush immediately so
// proxy timeouts never trigger and tokens render live in the browser.
if (p === '/api/ask/stream' && req.method === 'POST') {
const b = await body(req);
const q = String(b.q || '').trim();
if (!q) return json(res, { error: 'empty' }, 400);
// resolve/create the chat; history comes from the store (server-authoritative)
const cid = (b.chatId && chats.getChat(b.chatId)) ? b.chatId : chats.createChat().id;
const history = chats.historyFor(cid);
res.writeHead(200, { 'Content-Type': 'application/x-ndjson; charset=utf-8', 'Cache-Control': 'no-cache, no-transform', 'X-Accel-Buffering': 'no', 'Connection': 'keep-alive' });
const send = (o) => { try { res.write(JSON.stringify(o) + '\n'); } catch {} };
// "记住 X" → save a background fact, no AI call
{ const fact = extractRemember(q); if (fact) { const n = addNote(fact); const ans = n ? `✅ 记住了：${fact}\n\n（以后回答都会考虑这条。）` : '这条太短，没记下来。'; send({ type: 'meta', hits: [], chatId: cid }); send({ type: 'token', t: ans }); await persistExchange(cid, q, ans, [], []); send({ type: 'done', cited: [], chatId: cid }); return res.end(); } }
// aggregation/census question ("analyze ALL apologies") → exhaustive enumeration, not top-k RAG
const census = detectCensus(q);
if (census) {
send({ type: 'meta', hits: [], chatId: cid });
const collected = collectCensus(S.store.messages, census);
try {
const r = await censusAnalyze(census, collected, assembleCtx(), { onProgress: (msg) => send({ type: 'progress', msg }), onToken: (t) => send({ type: 'token', t }) });
await persistExchange(cid, q, r.report, [], []);
send({ type: 'done', cited: [], chatId: cid });
} catch (e) { send({ type: 'error', message: String(e.message || e) }); }
return res.end();
}
const retrievalQ = [...history.filter((h) => h.role === 'user').slice(-2).map((h) => h.content), q].join(' ');
const deep = isDeep(q);
const hits = corpus.search(retrievalQ, deep ? 16 : 12);
const hitMeta = hits.map((h, i) => ({ i, date: h.meta.date, dateEnd: h.meta.dateEnd, text: h.text.slice(0, 240), score: h.score }));
send({ type: 'meta', hits: hitMeta, chatId: cid });
try {
const { answer, cited } = dataMode()
? await askWithData(q, hits, assembleCtx(), history, { onToken: (t) => send({ type: 'token', t }), onProgress: (msg) => send({ type: 'progress', msg }), deep })
: await askAssistantStream(q, hits, assembleCtx(), history, (t) => send({ type: 'token', t }), deep);
await persistExchange(cid, q, answer, hitMeta, cited);
send({ type: 'done', cited, chatId: cid });
} catch (e) { send({ type: 'error', message: String(e.message || e) }); }
return res.end();
}
// draft a reply: scope=main → the space's main conversation; scope=other&chat=<name> → that chat only
if (p === '/api/reply-preview') {
const scope = u.searchParams.get('scope') || 'main';
const chat = u.searchParams.get('chat') || '';
const isMain = scope === 'main';
const pool = isMain ? S.store.messages : S.store.messages.filter((m) => !chat || m.sessionName === chat);
const tail = pool.slice(-16);
const label = isMain ? (CFG.target.name || CFG.name || '对方') : (chat || '对方');
if (!tail.length) return json(res, { error: '没有可参考的对话' }, 400);
const q = tail.slice(-6).map((m) => m.content).join(' ');
const ragSnippets = isMain ? corpus.search(q, 6) : [];
try {
const d = await draftReply({ label, isMain, tail, ragSnippets });
return json(res, { ...d, basedOn: ragSnippets.map((s) => ({ date: s.meta.date, who: s.meta.who, text: s.text.slice(0, 120) })) });
} catch (e) { return json(res, { error: e.message }, 500); }
}
if (p.startsWith('/api/reminders/') && p.endsWith('/toggle') && req.method === 'POST') {
markDone(decodeURIComponent(p.slice('/api/reminders/'.length, -'/toggle'.length)), Date.now());
broadcast('reminders', listReminders()); return json(res, { ok: true });
}
if (p === '/api/reminders/clear-done' && req.method === 'POST') { clearDone(); broadcast('reminders', listReminders()); return json(res, { ok: true }); }
const file = p === '/' ? '/index.html' : p;
const full = join(PUB, file.replace(/\.\./g, ''));
if (!full.startsWith(PUB) || full.startsWith(join(PUB, 'hub'))) { res.writeHead(404); return res.end('not found'); }
if (existsSync(full) && extname(full)) return serveFile(res, full);
res.writeHead(404); res.end('not found');
});
// ---- boot ----
async function main() {
loadReminders();
const retired = retireSensitive(Date.now()); // 敏感点 now live in the Living State memory
if (retired) log(`retired ${retired} sensitive todos -> Living State`);
loadLiving(); // evolving Living State memory
if (living.narrative) narrative = living.narrative;
if (living.observation) { S.store.ai.insight = living.observation; S.store.ai.insightAt = living.at; }
const BIZ = IS_BIZLIKE;
const persisted = S.loadPersisted(); // messages captured by earlier runs (messages.jsonl)
S.store.messages.sort((a, b) => a.ts - b.ts);
log(`loaded ${persisted} persisted messages`);
buildCorpus();
initMcp({ store: S.store, corpus: () => corpus, sessions: SESSIONS, living, activeMemory, listReminders, listNotes });
sources = createSources({
CFG, S, corpus: { add: (d) => corpus.add(d) }, broadcast, log, dataDir: DATA_DIR, extraTypes: await pluginConnectorTypes(),
onMessage: (m) => {
broadcast('message', m); broadcast('stats', S.snapshot());
if (m._tag?.care && !BIZ) broadcast('care', { at: m.ts, text: m.content.slice(0, 80), dir: m._tag.dir });
if (BIZ) bumpGroundSoon(); else bumpSummaryOnHeat(m._tag);
},
});
await sources.startAll();
buildCorpus();
// core tools the assistant can call
registerTool({
name: 'list_open_todos', readOnly: true,
description: '列出当前未完成的待办/承诺/需要回复的事项（已排序）。问到"我还有什么没做/待办/承诺"时用。',
progress: () => '读取待办清单…',
run: async () => { const r = listReminders(); return { open: (r.topRanked || r.open || []).slice(0, 20).map((t) => ({ text: t.text, kind: t.kind, score: t.score })) }; },
});
// outbound messages from the assistant are side-effectful → approval queue (readOnly:false)
registerTool({
name: 'send_message', readOnly: false,
description: '通过已连接的平台给某个对话发送一条消息（需要用户在面板批准后才会真正发送）。只在用户明确要求发送时使用。',
parameters: { type: 'object', properties: { chatId: { type: 'string', description: '对话 id（list 里的 session）' }, text: { type: 'string' } }, required: ['chatId', 'text'] },
preview: (a) => `发送到 ${NAMES[a.chatId] || a.chatId}：${String(a.text || '').slice(0, 120)}`,
progress: () => '准备发送…',
run: (a) => sources.send({ chatId: a.chatId, text: a.text }),
});
setExecutor(executeTool); // approval queue runs actions through the tool registry once approved
plugins = await activatePlugins({
CFG, log, broadcast, registerTool, dataDir: DATA_DIR,
store: S.store, messages: () => S.store.messages, living: () => living, reminders: () => listReminders(),
send: (chatId, text, connector) => sources.send({ chatId, text, connector }),
notify: (title, text, opts = {}) => sendPush(title, text, { click: CFG.notify?.dashboardUrl, ...opts }),
addApproval,
});
setInterval(() => broadcast('stats', S.snapshot()), 4000);
setInterval(() => S.saveSnapshotFile(), 60000);
if (CFG.ai.enabled) {
// todos + Living State run for every template (extract, score, auto-reconcile "what's done")
armSummary(12000);
setTimeout(analysisTick, 8000); setInterval(analysisTick, CFG.ai.analyzeIntervalMs || 20000); // per-message sentiment; viewer-gated
setInterval(reminderTick, CFG.ai.reminderIntervalMs);
setInterval(reconcileTick, CFG.ai.reconcileIntervalMs || CFG.ai.reminderIntervalMs);
setTimeout(reminderTick, 22000);
setTimeout(reconcileTick, 70000);
if (!BIZ) {
// relationship-only: twice-daily re-grounding + 档案/视角 + 关系体检
setInterval(narrativeTick, 12 * 3600000);
if (!living.groundedAt || Date.now() - living.groundedAt > 12 * 3600000) setTimeout(narrativeTick, 35000);
if (!regen || Date.now() - regen.at > 20 * 3600000) setTimeout(regenTick, 45000);
if (!checkup || Date.now() - (checkup.at || 0) > 20 * 3600000) setTimeout(checkupTick, 90000);
setTimeout(() => { regenTick(); setTimeout(checkupTick, 60000); setInterval(() => { regenTick(); setTimeout(checkupTick, 60000); }, 24 * 3600000); }, msUntil(20));
} else {
// multi-chat: full per-chat grounding builds the 概览 board — on boot (if stale) + daily
if (!living.at || Date.now() - living.at > 20 * 3600000) setTimeout(groundBizTick, 6000);
setTimeout(() => { groundBizTick(); setInterval(groundBizTick, 24 * 3600000); }, msUntil(6));
}
}
// loopback only by default; set "host": "0.0.0.0" in the space config to expose it on your network
server.listen(CFG.port, CFG.host || '127.0.0.1', () => log(`\n space "${CFG.name || CFG.id || CONTEXT_TYPE}" -> http://${CFG.host || '127.0.0.1'}:${CFG.port}\n history ${S.store.messages.length} msgs · template ${CONTEXT_TYPE} · AI ${CFG.ai.enabled ? CFG.ai.model : 'off'}\n`));
}
process.on('unhandledRejection', (e) => log('unhandledRejection', String(e)));
main();