clawmates/handoff/realtime/server.mjs

// Clawmates Live Visualization Bridge
// ------------------------------------
// Dependency-free Node (>=20) service. Two modes:
//   CLAWMATES_MODE=demo  → emits synthetic events from the taxonomy so the viz lights up offline.
//   CLAWMATES_MODE=live  → subscribes to the cm-api run SSE stream, normalizes via normalize(),
//                          and re-emits one fan-out SSE feed for browsers.
//
// It is READ-ONLY on the upstream stream. It never holds or forwards secret-broker credentials,
// and it exposes NO door-approval endpoint. Approvals stay on the authenticated cm-api path.
//
// Run:  node server.mjs        (reads ../.env if present, plus process env)

import http from 'node:http';
import fs from 'node:fs';
import path from 'node:path';
import { fileURLToPath } from 'node:url';

// ---- tiny .env loader (no dependency) -------------------------------------
const __dir = path.dirname(fileURLToPath(import.meta.url));
for (const envPath of [path.join(__dir, '..', '.env'), path.join(__dir, '.env')]) {
  try {
    for (const line of fs.readFileSync(envPath, 'utf8').split('\n')) {
      const m = line.match(/^\s*([A-Z0-9_]+)\s*=\s*(.*)\s*$/);
      if (m && process.env[m[1]] === undefined) process.env[m[1]] = m[2].replace(/^["']|["']$/g, '');
    }
  } catch { /* no .env, that's fine */ }
}

const MODE = process.env.CLAWMATES_MODE || 'demo';
const PORT = parseInt(process.env.LIVE_PORT || '8420', 10);
const ALLOWED = (process.env.LIVE_ALLOWED_ORIGINS || 'http://localhost:8080')
  .split(',').map(s => s.trim()).filter(Boolean);

// ---- fan-out hub -----------------------------------------------------------
const clients = new Set();             // res objects
const lastState = new Map();           // key → last event (for replay-on-connect)

function keyFor(type, d) {
  if (type === 'agent.status' || type === 'agent.task.update') return type + ':' + d.agentId;
  if (type === 'node.activity') return type + ':' + d.nodeId;
  if (type === 'telemetry' || type === 'topology.update') return type;
  if (type === 'routine.update') return type + ':' + d.routineId;
  return null; // streaming/ephemeral events (deltas, touches, messages) are not replayed
}

function broadcast(type, data) {
  const frame = `event: ${type}\ndata: ${JSON.stringify(data)}\n\n`;
  const k = keyFor(type, data);
  if (k) lastState.set(k, frame);
  for (const res of clients) res.write(frame);
}

// ---- HTTP server: /live (SSE), /healthz -----------------------------------
const server = http.createServer((req, res) => {
  const origin = req.headers.origin;
  const cors = origin && ALLOWED.includes(origin) ? origin : ALLOWED[0] || '*';

  if (req.url === '/healthz') {
    res.writeHead(200, { 'content-type': 'application/json' });
    return res.end(JSON.stringify({ ok: true, mode: MODE, clients: clients.size }));
  }

  if (req.url && req.url.startsWith('/live')) {
    res.writeHead(200, {
      'content-type': 'text/event-stream',
      'cache-control': 'no-cache, no-transform',
      'connection': 'keep-alive',
      'access-control-allow-origin': cors,
      'x-accel-buffering': 'no',          // disable nginx buffering
    });
    res.write('retry: 2000\n\n');         // browser auto-reconnect after 2s
    // replay current state so a late view paints immediately (mirrors checkpoint resume)
    for (const frame of lastState.values()) res.write(frame);
    clients.add(res);
    const ka = setInterval(() => res.write(': keepalive\n\n'), 15000);
    req.on('close', () => { clearInterval(ka); clients.delete(res); });
    return;
  }

  res.writeHead(404); res.end('not found');
});

server.listen(PORT, () => {
  console.log(`Live bridge on http://localhost:${PORT}/live  (SSE)   ·  mode=${MODE}`);
  if (MODE === 'demo') startDemo();
  else startLive();
});

// ===========================================================================
// LIVE MODE — subscribe to cm-api run SSE, normalize to taxonomy
// ===========================================================================
async function startLive() {
  const url = process.env.CM_API_SSE_URL;
  const token = process.env.CM_API_TOKEN;
  if (!url) { console.error('CM_API_SSE_URL is required in live mode'); process.exit(1); }

  console.log(`[live] subscribing to ${url}`);
  // Native fetch streaming (Node 20+). Reconnects with backoff.
  let backoff = 1000;
  for (;;) {
    try {
      const resp = await fetch(url, { headers: token ? { authorization: `Bearer ${token}` } : {} });
      if (!resp.ok || !resp.body) throw new Error('upstream ' + resp.status);
      backoff = 1000;
      const reader = resp.body.getReader();
      const dec = new TextDecoder();
      let buf = '';
      for (;;) {
        const { value, done } = await reader.read();
        if (done) break;
        buf += dec.decode(value, { stream: true });
        const frames = buf.split('\n\n'); buf = frames.pop() || '';
        for (const f of frames) handleUpstreamFrame(f);
      }
    } catch (e) {
      console.error('[live] stream error:', e.message, '— retrying in', backoff, 'ms');
      await new Promise(r => setTimeout(r, backoff));
      backoff = Math.min(backoff * 2, 30000);
    }
  }
}

function handleUpstreamFrame(frame) {
  // upstream frame: "event: <t>\ndata: <json>"
  let evt = 'message', data = '';
  for (const line of frame.split('\n')) {
    if (line.startsWith('event:')) evt = line.slice(6).trim();
    else if (line.startsWith('data:')) data += line.slice(5).trim();
  }
  let parsed; try { parsed = JSON.parse(data); } catch { return; }
  for (const [type, payload] of normalize(evt, parsed)) broadcast(type, payload);
}

// THE INTEGRATION SEAM ------------------------------------------------------
// Translate one raw runner event into zero+ taxonomy events. Extend this map as
// you learn the runner's event shapes. Start with the three high-value ones.
function* normalize(evt, p) {
  switch (evt) {
    case 'turn.started':
      yield ['agent.status', { agentId: p.agentId, status: 'working', role: p.role }];
      if (p.task) yield ['agent.task.update', {
        agentId: p.agentId, taskId: p.taskId, title: p.task.title,
        elapsedMs: p.task.elapsedMs || 0, steps: p.task.steps || [],
      }];
      break;
    case 'turn.token':                    // streamed reasoning
      yield ['agent.reasoning.delta', { agentId: p.agentId, text: p.text, channel: p.channel || 'think' }];
      break;
    case 'tool.invoked':
      yield ['agent.tool.call', { agentId: p.agentId, tool: p.tool, target: p.target, doorRequired: !!p.doorRequired }];
      // a tool that leaves the sandbox surfaces as a touch on the world graph
      if (p.nodeId) yield ['world.touch', { agentId: p.agentId, nodeId: p.nodeId, kind: p.nodeKind || 'service' }];
      break;
    case 'door.requested':
      yield ['door.request', { doorId: p.doorId, agentId: p.agentId, action: p.action, target: p.target, summary: p.summary }];
      break;
    case 'door.resolved':
      yield ['door.resolve', { doorId: p.doorId, decision: p.decision, by: p.by }];
      break;
    case 'agent.message':
      yield ['agent.message', { fromAgentId: p.from, toAgentId: p.to, text: p.text, ts: p.ts }];
      break;
    case 'runner.telemetry':
      yield ['telemetry', { tokensPerMin: p.tokensPerMin, costPerHr: p.costPerHr, loops: p.loops, doorsPending: p.doorsPending }];
      break;
    case 'routine.tick':
      yield ['routine.update', { routineId: p.routineId, name: p.name, kind: p.kind, owner: p.owner, schedule: p.schedule, progress: p.progress, state: p.state }];
      break;
    // unknown events are ignored — the bridge is forward-compatible.
  }
}

// ===========================================================================
// DEMO MODE — synthetic world activity so everything lights up with no backend
// ===========================================================================
function startDemo() {
  const agents = ['morpheus', 'smith'];
  const nodes = [
    ['runtime','cm-runtime','service'], ['pr214','PR #214','event'], ['advdb','advisory-db','service'],
    ['slack','#eng','event'], ['orch','cm-orch','service'], ['deploy','PROD deploy','event'],
    ['docs','spec §15','service'], ['bench','topo-bench','service'], ['broker','secret-broker','service'],
  ];
  // seed topology + status
  broadcast('topology.update', {
    formation: 'live',
    nodes: nodes.map(([id, label, kind]) => ({ id, tier: kind, label })),
  });
  agents.forEach(a => broadcast('agent.status', { agentId: a, status: 'working', role: a === 'morpheus' ? 'Project Manager' : 'Research Specialist' }));

  // agents converge on random nodes (Gource)
  setInterval(() => {
    const a = agents[Math.random() * agents.length | 0];
    const [id, label, kind] = nodes[Math.random() * nodes.length | 0];
    broadcast('world.touch', { agentId: a, nodeId: id, kind });
    broadcast('node.activity', { nodeId: id, label, kind, heat: 0.6 + Math.random() * 0.4 });
  }, 700);

  // reasoning + comms
  const thoughts = [
    'checking lifetime on &\'a mut Guard across await',
    'cargo clippy --workspace clean',
    'this holds a std::sync::Mutex across .await — fix it',
    'cloning under a scoped lock, dropping the guard first',
  ];
  setInterval(() => broadcast('agent.reasoning.delta', { agentId: 'morpheus', text: thoughts[Math.random() * thoughts.length | 0] + ' … ' }), 1600);
  setInterval(() => broadcast('agent.message', { fromAgentId: 'morpheus', toAgentId: 'smith', text: 'can you audit the crate tree before I approve?', ts: new Date().toISOString() }), 5200);

  // telemetry drift
  let tok = 38000;
  setInterval(() => {
    tok += (Math.random() - 0.5) * 4000;
    broadcast('telemetry', { tokensPerMin: Math.round(tok), costPerHr: 0.42, loops: 3, doorsPending: 1 });
  }, 2000);

  // an occasional door
  setInterval(() => {
    broadcast('door.request', { doorId: 'd' + Date.now(), agentId: 'morpheus', action: 'github.review.comment', target: 'PR #214', summary: 'Post code review on PR #214' });
  }, 12000);

  console.log('[demo] emitting synthetic events — open a viz with ?live=http://localhost:' + PORT + '/live');
}