From 34fb1743082c7e30a74850121ef685339fd8b3fb Mon Sep 17 00:00:00 2001 From: artheru Date: Thu, 11 Jun 2026 02:39:33 +0800 Subject: [PATCH] api: native chat + no-cache file relay (chat backend, iter 2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - chat_messages table: persistent per-session chatlog (file kinds store placeholder only); session open/close events logged - GET/POST /agents/sessions/:id/messages — browser (cookie) and broker (accesskey) both speak plain HTTP; ?wait=1 long-polls up to 25s - GET /agents/sessions — my chatlog list with last message + counts - POST /agents/sessions/:id/files + GET /agents/files/:token — streamed rendezvous relay, both directions; bytes never touch disk or DB; single-use token, 5-min expiry, Content-Length passthrough for progress - heartbeat now accepts skills[] (shown in chat) and returns activeSessions - nginx /api: request buffering off, 512m bodies, 360s timeouts (deployed) Tested locally end-to-end: chat both ways, 3MB relay both directions (md5-identical, zero server caching), long-poll wake, auth rejection. Co-Authored-By: Claude Fable 5 --- server/amerc-api.mjs | 162 +++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 156 insertions(+), 6 deletions(-) diff --git a/server/amerc-api.mjs b/server/amerc-api.mjs index 84eebdd..57a679b 100644 --- a/server/amerc-api.mjs +++ b/server/amerc-api.mjs @@ -58,6 +58,14 @@ db.exec(` CREATE TABLE IF NOT EXISTS agent_sessions ( id INTEGER PRIMARY KEY AUTOINCREMENT, instance_id INTEGER NOT NULL, class_id INTEGER, party TEXT DEFAULT '', connector TEXT DEFAULT 'ws', status TEXT NOT NULL DEFAULT 'active', started_at INTEGER NOT NULL, ended_at INTEGER); + -- chatlog: every session keeps its messages. kind 'file' stores only a + -- placeholder (name/size/mime) — bytes relay through memory, never stored. + CREATE TABLE IF NOT EXISTS chat_messages ( + id INTEGER PRIMARY KEY AUTOINCREMENT, session_id INTEGER NOT NULL, seq INTEGER NOT NULL, + role TEXT NOT NULL DEFAULT 'user', kind TEXT NOT NULL DEFAULT 'text', content TEXT DEFAULT '', + file_name TEXT DEFAULT '', file_size INTEGER DEFAULT 0, file_mime TEXT DEFAULT '', file_token TEXT DEFAULT '', + created_at INTEGER NOT NULL); + CREATE INDEX IF NOT EXISTS idx_chat_session_seq ON chat_messages(session_id, seq); CREATE TABLE IF NOT EXISTS class_subscriptions ( id INTEGER PRIMARY KEY AUTOINCREMENT, class_id INTEGER NOT NULL, party TEXT NOT NULL, token TEXT NOT NULL, created_at INTEGER NOT NULL); @@ -67,6 +75,60 @@ db.exec(` `); try { db.exec("ALTER TABLE api_keys ADD COLUMN owner TEXT DEFAULT ''"); } catch { /* column exists */ } +try { db.exec("ALTER TABLE agent_instances ADD COLUMN skills TEXT DEFAULT ''"); } catch { /* column exists */ } + +// ---- live chat plumbing (in-memory; survives nothing, stores nothing) ------- +// waiters: long-poll watchers per session, woken when a message lands. +const chatWaiters = new Map(); // sessionId -> Set +function wakeChat(sessionId) { + const set = chatWaiters.get(sessionId); + if (!set) return; + chatWaiters.delete(sessionId); + for (const fn of set) { try { fn(); } catch { /* waiter gone */ } } +} +function waitChat(sessionId, ms) { + return new Promise((resolve) => { + let set = chatWaiters.get(sessionId); + if (!set) { set = new Set(); chatWaiters.set(sessionId, set); } + const done = () => { clearTimeout(t); set.delete(fn); resolve(); }; + const fn = done; + set.add(fn); + const t = setTimeout(done, ms); + }); +} +// file rendezvous: producer's request stream is piped to the consumer's +// response when it claims the token. The server never writes the bytes. +const FILE_CLAIM_MS = 5 * 60 * 1000; +const fileRelays = new Map(); // token -> { req, res, meta, timer, claimed } +function offerFile(token, req, res, meta) { + req.pause(); + const entry = { req, res, meta, claimed: false }; + entry.timer = setTimeout(() => { + if (!entry.claimed && fileRelays.get(token) === entry) { + fileRelays.delete(token); + send(res, 408, { ok: false, error: 'file not claimed within 5 minutes' }); + } + }, FILE_CLAIM_MS); + fileRelays.set(token, entry); + req.on('close', () => { if (!entry.claimed && fileRelays.get(token) === entry) { clearTimeout(entry.timer); fileRelays.delete(token); } }); +} +function claimFile(token, consumerRes) { + const entry = fileRelays.get(token); + if (!entry || entry.claimed) return false; + entry.claimed = true; + clearTimeout(entry.timer); + fileRelays.delete(token); + const { req: prodReq, res: prodRes, meta } = entry; + const headers = { 'Content-Type': meta.mime || 'application/octet-stream', 'Content-Disposition': `attachment; filename="${encodeURIComponent(meta.name || 'file')}"`, 'Cache-Control': 'no-store' }; + if (meta.size) headers['Content-Length'] = meta.size; + consumerRes.writeHead(200, headers); + prodReq.pipe(consumerRes); + prodReq.on('end', () => send(prodRes, 200, { ok: true, delivered: true })); + prodReq.on('error', () => { try { consumerRes.destroy(); } catch { /* gone */ } }); + consumerRes.on('close', () => { if (!consumerRes.writableEnded) try { prodReq.destroy(); send(prodRes, 499, { ok: false, error: 'receiver disconnected' }); } catch { /* gone */ } }); + prodReq.resume(); + return true; +} const SERVER_IP = process.env.AMERC_SERVER_IP || '104.168.145.233'; // edge token: signed {host} so the showcase-edge can verify a broker may bind that host @@ -335,7 +397,23 @@ const server = http.createServer(async (req, res) => { if (age > INSTANCE_LOST_MS && inst.status !== 'pending') return 'lost'; return inst.status; }; - const instView = (inst, key = false) => ({ id: inst.id, class_id: inst.class_id, owner: inst.owner, name: inst.name, visibility: inst.visibility, kind: inst.kind, status: liveStatus(inst), broker: inst.broker, last_heartbeat: inst.last_heartbeat, created_at: inst.created_at, ...(key ? { accesskey: inst.accesskey } : {}) }); + const parseSkills = (s) => { try { const v = JSON.parse(s || '[]'); return Array.isArray(v) ? v : []; } catch { return []; } }; + const instView = (inst, key = false) => ({ id: inst.id, class_id: inst.class_id, owner: inst.owner, name: inst.name, visibility: inst.visibility, kind: inst.kind, status: liveStatus(inst), broker: inst.broker, skills: parseSkills(inst.skills), last_heartbeat: inst.last_heartbeat, created_at: inst.created_at, ...(key ? { accesskey: inst.accesskey } : {}) }); + // chatlog insert: persists the message (file kinds keep only the placeholder) and wakes long-pollers + const addMsg = (sessionId, role, kind, content, file = {}) => { + const seq = db.prepare('SELECT COALESCE(MAX(seq),0) s FROM chat_messages WHERE session_id=?').get(sessionId).s + 1; + db.prepare('INSERT INTO chat_messages(session_id,seq,role,kind,content,file_name,file_size,file_mime,file_token,created_at) VALUES(?,?,?,?,?,?,?,?,?,?)') + .run(sessionId, seq, role, kind, String(content || '').slice(0, 32000), String(file.name || '').slice(0, 200), Number(file.size) || 0, String(file.mime || '').slice(0, 100), file.token || '', Date.now()); + wakeChat(sessionId); + return seq; + }; + // who may touch a session: the instance's broker (accesskey) acts as 'agent'; + // the hiring party / instance owner / admin act as 'user'. + const sessionRole = (sess, inst, akey) => { + if (akey && akey === inst.accesskey) return 'agent'; + if (id && (sess.party === who(id) || inst.owner === who(id) || isAdmin)) return 'user'; + return null; + }; const slugify = (s) => String(s || 'agent').toLowerCase().replace(/[^a-z0-9]+/g, '-').replace(/^-|-$/g, '').slice(0, 40) || 'agent'; // public stats for the tavern frontpage @@ -442,8 +520,11 @@ const server = http.createServer(async (req, res) => { const b = await readBody(req, MAX_UPLOAD) || {}; const inst = db.prepare('SELECT * FROM agent_instances WHERE id=?').get(Number(m[1])); if (!inst) return send(res, 404, { ok: false, error: 'not found' }); if (b.accesskey !== inst.accesskey) return send(res, 401, { ok: false, error: 'bad accesskey' }); const status = b.status === 'down' ? 'down' : 'online'; - db.prepare('UPDATE agent_instances SET status=?,last_heartbeat=?,broker=?,tui=? WHERE id=?').run(status, Date.now(), b.broker || inst.broker, b.tui != null ? String(b.tui).slice(0, 20000) : inst.tui, inst.id); - return send(res, 200, { ok: true, status }); + const skills = Array.isArray(b.skills) ? JSON.stringify(b.skills.slice(0, 64).map((s) => String(s).slice(0, 80))) : inst.skills; + db.prepare('UPDATE agent_instances SET status=?,last_heartbeat=?,broker=?,tui=?,skills=? WHERE id=?').run(status, Date.now(), b.broker || inst.broker, b.tui != null ? String(b.tui).slice(0, 20000) : inst.tui, skills, inst.id); + // tell the broker which of its sessions are live, so it can long-poll their chat + const live = db.prepare("SELECT id FROM agent_sessions WHERE instance_id=? AND status='active'").all(inst.id).map((r) => r.id); + return send(res, 200, { ok: true, status, activeSessions: live }); } if ((m = path.match(/^\/agents\/instances\/(\d+)$/)) && method === 'DELETE') { const inst = db.prepare('SELECT * FROM agent_instances WHERE id=?').get(Number(m[1])); if (!inst) return send(res, 404, { ok: false, error: 'not found' }); @@ -458,11 +539,80 @@ const server = http.createServer(async (req, res) => { const b = await readBody(req) || {}; const authed = owns(inst) || (b.accesskey && b.accesskey === inst.accesskey) || (inst.visibility === 'public' && id); if (!authed) return send(res, 403, { ok: false, error: 'not authorized — need login (public) or accesskey (private)' }); const info = db.prepare('INSERT INTO agent_sessions(instance_id,class_id,party,connector,status,started_at) VALUES(?,?,?,?,?,?)') - .run(inst.id, inst.class_id, id ? who(id) : 'anon', b.connector || 'ws', 'active', Date.now()); - return send(res, 200, { ok: true, sessionId: Number(info.lastInsertRowid), relayToken: inst.accesskey, relayWs: `wss://amerc.ai/webagent/ws?role=iframe&token=${inst.accesskey}`, chatboxJs: `https://webagent.amerc.ai/client.js?token=${inst.accesskey}` }); + .run(inst.id, inst.class_id, id ? who(id) : 'anon', b.connector || 'chat', 'active', Date.now()); + const sid = Number(info.lastInsertRowid); + addMsg(sid, 'system', 'event', `session opened by ${id ? who(id) : 'anon'}`); + return send(res, 200, { ok: true, sessionId: sid, relayToken: inst.accesskey, relayWs: `wss://amerc.ai/webagent/ws?role=iframe&token=${inst.accesskey}`, chatboxJs: `https://webagent.amerc.ai/client.js?token=${inst.accesskey}` }); } if ((m = path.match(/^\/agents\/sessions\/(\d+)\/close$/)) && method === 'POST') { - db.prepare("UPDATE agent_sessions SET status='closed', ended_at=? WHERE id=?").run(Date.now(), Number(m[1])); return send(res, 200, { ok: true }); + db.prepare("UPDATE agent_sessions SET status='closed', ended_at=? WHERE id=?").run(Date.now(), Number(m[1])); + addMsg(Number(m[1]), 'system', 'event', 'session closed'); + return send(res, 200, { ok: true }); + } + + // ---------- CHAT: native amerc chat — no external relay needed ---------- + // my chatlog: every session I started, or that runs on an instance I own + if (path === '/agents/sessions' && method === 'GET') { + if (!id) return needAuth(); + const me = who(id); + const rows = db.prepare(`SELECT s.*, c.name class_name, c.slug class_slug, i.name inst_name, i.owner inst_owner + FROM agent_sessions s LEFT JOIN agent_classes c ON c.id=s.class_id LEFT JOIN agent_instances i ON i.id=s.instance_id + WHERE s.party=? OR i.owner=?${isAdmin ? ' OR 1=1' : ''} ORDER BY s.started_at DESC LIMIT 100`).all(me, me); + const out = rows.map((s) => { + const last = db.prepare('SELECT role,kind,content,file_name,created_at FROM chat_messages WHERE session_id=? ORDER BY seq DESC LIMIT 1').get(s.id); + const n = db.prepare("SELECT COUNT(*) n FROM chat_messages WHERE session_id=? AND kind!='event'").get(s.id).n; + return { ...s, messageCount: n, last }; + }); + return send(res, 200, { ok: true, sessions: out }); + } + // poll/long-poll messages: GET ?after=&wait=1 — agent side passes ?accesskey= + if ((m = path.match(/^\/agents\/sessions\/(\d+)\/messages$/)) && method === 'GET') { + const sess = db.prepare('SELECT * FROM agent_sessions WHERE id=?').get(Number(m[1])); if (!sess) return send(res, 404, { ok: false, error: 'not found' }); + const inst = db.prepare('SELECT * FROM agent_instances WHERE id=?').get(sess.instance_id); + const role = sessionRole(sess, inst, url.searchParams.get('accesskey')); + if (!role) return send(res, 403, { ok: false, error: 'not in this session' }); + const after = Number(url.searchParams.get('after') || 0); + const fetchRows = () => db.prepare('SELECT seq,role,kind,content,file_name,file_size,file_mime,file_token,created_at FROM chat_messages WHERE session_id=? AND seq>? ORDER BY seq LIMIT 200').all(sess.id, after); + let rows = fetchRows(); + if (!rows.length && url.searchParams.get('wait')) { await waitChat(sess.id, 25000); rows = fetchRows(); } + const fresh = db.prepare('SELECT status FROM agent_sessions WHERE id=?').get(sess.id); + return send(res, 200, { ok: true, messages: rows, sessionStatus: fresh.status, instanceStatus: liveStatus(inst), skills: parseSkills(inst.skills) }); + } + // send a message: browser (cookie) or broker (accesskey in body) + if ((m = path.match(/^\/agents\/sessions\/(\d+)\/messages$/)) && method === 'POST') { + const sess = db.prepare('SELECT * FROM agent_sessions WHERE id=?').get(Number(m[1])); if (!sess) return send(res, 404, { ok: false, error: 'not found' }); + if (sess.status !== 'active') return send(res, 409, { ok: false, error: 'session closed' }); + const inst = db.prepare('SELECT * FROM agent_instances WHERE id=?').get(sess.instance_id); + const b = await readBody(req) || {}; + const role = sessionRole(sess, inst, b.accesskey); + if (!role) return send(res, 403, { ok: false, error: 'not in this session' }); + if (!b.content || typeof b.content !== 'string') return send(res, 400, { ok: false, error: 'content required' }); + const seq = addMsg(sess.id, role, 'text', b.content); + return send(res, 200, { ok: true, seq }); + } + // FILE RELAY (no cache): producer streams the raw body here; the bytes wait + // in this paused request until the other side claims the token — the server + // stores nothing. The chatlog records only "there was a file". + if ((m = path.match(/^\/agents\/sessions\/(\d+)\/files$/)) && method === 'POST') { + const sess = db.prepare('SELECT * FROM agent_sessions WHERE id=?').get(Number(m[1])); if (!sess) return send(res, 404, { ok: false, error: 'not found' }); + if (sess.status !== 'active') return send(res, 409, { ok: false, error: 'session closed' }); + const inst = db.prepare('SELECT * FROM agent_instances WHERE id=?').get(sess.instance_id); + const role = sessionRole(sess, inst, url.searchParams.get('accesskey')); + if (!role) return send(res, 403, { ok: false, error: 'not in this session' }); + const meta = { + name: (url.searchParams.get('name') || 'file').replace(/[/\\]/g, '_').slice(0, 200), + size: Number(req.headers['content-length']) || Number(url.searchParams.get('size')) || 0, + mime: req.headers['content-type'] && req.headers['content-type'] !== 'application/x-www-form-urlencoded' ? req.headers['content-type'] : 'application/octet-stream', + }; + const token = 'f_' + crypto.randomBytes(18).toString('base64url'); + offerFile(token, req, res, meta); + addMsg(sess.id, role, 'file', `sent ${meta.name}`, { ...meta, token }); + return; // responds when claimed (delivered) or after the 5-min timeout + } + // claim a relayed file: single-use capability token from the chat message + if ((m = path.match(/^\/agents\/files\/([A-Za-z0-9_-]+)$/)) && method === 'GET') { + if (claimFile(m[1], res)) return; // piped + return send(res, 410, { ok: false, error: 'file gone — relay links are single-use and expire in 5 minutes' }); } // ---------- self-service agent keys (any logged-in user) ----------