← back to Harness Viewer

server.js

217 lines

#!/usr/bin/env node
// Harness Cockpit — live viewer for Steve's agent-orchestration harness.
// Basic-auth admin/DW2024!  ·  dep-free (Node built-in http)  ·  live-scan + ~6s cache.
// Mirrors the architecture of ~/Projects/stack-map-viewer/server.js.
const http = require('http');
const https = require('https');
const { execSync } = require('child_process');
const fs = require('fs');
const os = require('os');
const path = require('path');

const HOME = os.homedir();
const EVENTS = path.join(HOME, '.claude/tickets/events.jsonl');
const PENDING = path.join(HOME, '.claude/yolo-queue/pending-approval');
const LEDGER = path.join(HOME, '.claude/yolo-queue/executed-reversible/ledger.jsonl');
const AUDIT_SCRIPT = path.join(HOME, 'Projects/ticket-system/scripts/audit-overnight-closes.mjs');
const AUTH = 'Basic ' + Buffer.from('admin:DW2024!').toString('base64');
const PORT = process.env.PORT || 9770;

// ── safe shell: 6s timeout, graceful empty on any failure ────────────────────
const sh = (c, ms = 6000) => { try { return execSync(c, { encoding: 'utf8', timeout: ms, maxBuffer: 32 * 1024 * 1024 }); } catch { return ''; } };
const shq = (s) => "'" + String(s).replace(/'/g, "'\\''") + "'";
const trunc = (s, n) => { s = String(s || ''); return s.length > n ? s.slice(0, n - 1) + '…' : s; };
const shortId = (id) => { const m = String(id || '').match(/^(TK-\d+)/); return m ? m[1] : String(id || ''); };

// Read the last N lines of a big file WITHOUT loading it whole. Never read
// events.jsonl (57MB) or ledger (86k+ lines) in full — tail + per-line parse.
function tailJSON(file, n) {
  const raw = sh(`tail -n ${n} ${shq(file)} 2>/dev/null`);
  if (!raw) return [];
  const out = [];
  for (const ln of raw.split('\n')) {
    const t = ln.trim();
    if (!t) continue;
    try { out.push(JSON.parse(t)); } catch { /* skip malformed line */ }
  }
  return out;
}

// ── ENGINES: probe the "full AI setup" (2s each) ─────────────────────────────
function probeHTTP(url, timeoutMs = 2000) {
  return new Promise((resolve) => {
    let done = false;
    const finish = (up) => { if (!done) { done = true; resolve(up); } };
    try {
      const lib = url.startsWith('https') ? https : http;
      const req = lib.get(url, (r) => { finish(r.statusCode === 200); r.resume(); });
      req.on('error', () => finish(false));
      req.setTimeout(timeoutMs, () => { req.destroy(); finish(false); });
    } catch { finish(false); }
  });
}
const ENGINE_DEFS = [
  { key: 'exo', label: 'exo ring', kind: 'http', endpoint: 'http://127.0.0.1:52415/' },
  { key: 'openclaw', label: 'openclaw MLX', kind: 'http', endpoint: 'http://127.0.0.1:8000/v1/models' },
  { key: 'gateway', label: 'ai-cluster gateway', kind: 'http', endpoint: 'http://127.0.0.1:8080/' },
  { key: 'codex', label: 'codex CLI', kind: 'exec', endpoint: '~/.local/bin/codex', test: path.join(HOME, '.local/bin/codex') },
  { key: 'askopenai', label: 'ask-openai', kind: 'exec', endpoint: '~/bin/ask-openai', test: path.join(HOME, 'bin/ask-openai') },
];
async function probeEngines() {
  const now = new Date().toISOString();
  const results = await Promise.all(ENGINE_DEFS.map(async (e) => {
    let up = false;
    if (e.kind === 'http') up = await probeHTTP(e.endpoint, 2000);
    else { try { fs.accessSync(e.test, fs.constants.X_OK); up = true; } catch { up = false; } }
    return { key: e.key, label: e.label, endpoint: e.endpoint, up, ts: now };
  }));
  return results;
}

// ── LIVE LANES + TICKETS: from a single tail of events.jsonl ─────────────────
function analyzeEvents() {
  const ev = tailJSON(EVENTS, 4000);
  const now = Date.now();
  const WINDOW = 30 * 60 * 1000; // 30 min
  const lanes = new Map();   // agent -> {agent, actions, lastText, lastTs}
  const tickets = new Map(); // shortId -> {id, agent, lastText, ts, state}

  for (const e of ev) {
    if (!e || !e.ts) continue;
    const t = Date.parse(e.ts);
    if (isNaN(t)) continue;
    const text = e.text || (e.status ? '[' + e.status + ']' : e.type || '');
    // --- lanes: last 30 min, grouped by agent ---
    if (e.agent && now - t <= WINDOW) {
      let ln = lanes.get(e.agent);
      if (!ln) { ln = { agent: e.agent, actions: 0, lastText: '', lastTs: 0 }; lanes.set(e.agent, ln); }
      ln.actions++;
      if (t >= ln.lastTs) { ln.lastTs = t; ln.lastText = trunc(text, 140); }
    }
    // --- tickets: reduce by short id, derive state ---
    if (e.id) {
      const sid = shortId(e.id);
      let tk = tickets.get(sid);
      if (!tk) { tk = { id: sid, fullId: e.id, agent: e.agent || '', lastText: '', ts: 0, state: 'DOING' }; tickets.set(sid, tk); }
      let state = tk.state;
      if (e.type === 'status') { const s = (e.status || '').toLowerCase(); if (s === 'done') state = 'DONE'; else if (s === 'blocked') state = 'BLOCKED'; else state = 'DOING'; }
      else if (e.type === 'new' || e.type === 'take') state = 'DOING';
      else if (e.type === 'comment' || e.type === 'action' || e.type === 'note' || e.type === 'log') state = 'DOING';
      if (t >= tk.ts) { tk.ts = t; tk.lastText = trunc(text, 160); tk.state = state; if (e.agent) tk.agent = e.agent; }
      else if (e.type === 'status') { tk.state = state; } // status is authoritative even if slightly older
    }
  }

  const laneArr = [...lanes.values()].sort((a, b) => b.lastTs - a.lastTs)
    .map(l => ({ agent: l.agent, actions: l.actions, lastText: l.lastText, ts: new Date(l.lastTs).toISOString() }));
  const tkArr = [...tickets.values()].sort((a, b) => b.ts - a.ts).slice(0, 25)
    .map(t => ({ id: t.id, agent: t.agent, lastText: t.lastText, ts: new Date(t.ts).toISOString(), state: t.state }));
  const doing = [...tickets.values()].filter(t => t.state === 'DOING').length;
  return { lanes: laneArr, tickets: tkArr, doing };
}

// ── GATE QUEUE ───────────────────────────────────────────────────────────────
function scanGateQueue() {
  let files = [];
  try { files = fs.readdirSync(PENDING).filter(f => f.endsWith('.md')); } catch { return { total: 0, p3: 0, p24: 0, p7d: 0, newest: [] }; }
  const now = Date.now();
  let p3 = 0, p24 = 0, p7d = 0;
  const withMtime = [];
  for (const f of files) {
    let m = 0;
    try { m = fs.statSync(path.join(PENDING, f)).mtimeMs; } catch { continue; }
    const h = (now - m) / 3.6e6;
    if (h >= 3) p3++;
    if (h >= 24) p24++;
    if (h >= 168) p7d++;
    withMtime.push({ f, m });
  }
  withMtime.sort((a, b) => b.m - a.m);
  const newest = withMtime.slice(0, 25).map(x => ({ f: x.f, ts: new Date(x.m).toISOString() }));
  return { total: files.length, p3, p24, p7d, newest };
}

// ── REVERSIBLE LEDGER ────────────────────────────────────────────────────────
function scanLedger() {
  const rows = tailJSON(LEDGER, 40);
  return rows.reverse().map(r => ({
    ts: r.ts || '', agent: r.agent || '', ticket: r.ticket || '',
    action: trunc(r.action, 220), undo_cmd: trunc(r.undo_cmd, 260),
  }));
}

// ── aggregate state (all panels except on-demand audit), ~6s cache ───────────
let CACHE = null, CACHE_TS = 0;
async function buildState() {
  if (CACHE && Date.now() - CACHE_TS < 6000) return CACHE;
  const engines = await probeEngines();
  const ev = analyzeEvents();
  const gate = scanGateQueue();
  const ledger = scanLedger();
  const enginesUp = engines.filter(e => e.up).length;
  CACHE = {
    ts: new Date().toISOString(),
    engines, enginesUp, enginesTotal: engines.length,
    lanes: ev.lanes, tickets: ev.tickets, doing: ev.doing,
    gate, ledger,
    hasAudit: (() => { try { fs.accessSync(AUDIT_SCRIPT); return true; } catch { return false; } })(),
  };
  CACHE_TS = Date.now();
  return CACHE;
}

// ── memo reader (path-jailed to PENDING) ─────────────────────────────────────
function readMemo(name) {
  if (!name || typeof name !== 'string') return { error: 'no file' };
  const resolved = path.resolve(PENDING, name);
  const jail = path.resolve(PENDING) + path.sep;
  if (!(resolved + path.sep).startsWith(jail) && resolved !== path.resolve(PENDING)) return { error: 'path escape rejected' };
  if (!resolved.startsWith(jail)) return { error: 'path escape rejected' };
  if (!resolved.endsWith('.md')) return { error: 'not a memo' };
  let txt = '';
  try { txt = fs.readFileSync(resolved, 'utf8'); } catch { return { error: 'not found' }; }
  const lines = txt.split('\n').slice(0, 60).join('\n');
  return { file: path.basename(resolved), lines };
}

// ── audit (on-demand only) ───────────────────────────────────────────────────
function runAudit() {
  try { fs.accessSync(AUDIT_SCRIPT); } catch { return { error: 'audit script not found', script: AUDIT_SCRIPT }; }
  const out = sh(`node ${shq(AUDIT_SCRIPT)} 2>&1`, 30000);
  return { script: AUDIT_SCRIPT, output: out || '(no output / timed out)' };
}

const PAGE = fs.readFileSync(path.join(__dirname, 'index.html'), 'utf8');

const server = http.createServer(async (req, res) => {
  if (req.headers.authorization !== AUTH) {
    res.writeHead(401, { 'WWW-Authenticate': 'Basic realm="harness-cockpit"' });
    return res.end('auth required');
  }
  const u = new URL(req.url, 'http://x');
  try {
    if (u.pathname === '/api/state') {
      const s = await buildState();
      res.writeHead(200, { 'content-type': 'application/json' });
      return res.end(JSON.stringify(s));
    }
    if (u.pathname === '/api/memo') {
      const out = readMemo(u.searchParams.get('f') || '');
      res.writeHead(out.error ? 400 : 200, { 'content-type': 'application/json' });
      return res.end(JSON.stringify(out));
    }
    if (u.pathname === '/api/audit') {
      const out = runAudit();
      res.writeHead(200, { 'content-type': 'application/json' });
      return res.end(JSON.stringify(out));
    }
  } catch (e) {
    res.writeHead(500, { 'content-type': 'application/json' });
    return res.end(JSON.stringify({ error: String(e && e.message || e) }));
  }
  res.writeHead(200, { 'content-type': 'text/html' });
  res.end(PAGE);
});

server.listen(PORT, () => console.log(`harness cockpit → http://127.0.0.1:${PORT}  (admin/DW2024!)`));