api: native chat + no-cache file relay (chat backend, iter 2)

- 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 <noreply@anthropic.com>

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<fn>
+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);
+      const skills = Array.isArray(b.skills) ? JSON.stringify(b.skills.slice(0, 64).map((s) => String(s).slice(0, 80))) : inst.skills;
-      return send(res, 200, { ok: true, status });
+      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());
+        .run(inst.id, inst.class_id, id ? who(id) : 'anon', b.connector || 'chat', '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}` });
+      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=<seq>&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) ----------