// agent-stream-hub — schema-agnostic relay.
//
// In-memory per-run entries. Sources register, push snapshots, heartbeat.
// Viewers list runs and fetch /latest (most-recent snapshot) and /history
// (older snapshots for backfill). No markers, no blocks — the viewer just
// stitches snapshots into a continuous buffer.
//
// Bind to 127.0.0.1 — public exposure is via cloudflared (separate process).

const http = require('http');
const fs = require('fs');
const path = require('path');
const os = require('os');
const crypto = require('crypto');
const zlib = require('zlib');
const { execFile } = require('child_process');

const HOST = '127.0.0.1';
const PORT = process.env.PORT === undefined ? 9000 : Number(process.env.PORT);

const RING_CAP = 400;             // entries kept per run (snapshots + markers)
const EVICT_MS = 24 * 60 * 60 * 1000; // forget runs after a day with no activity
const MAX_BODY = 4 * 1024 * 1024; // 4 MiB per POST
const MAX_RUNS = 128;
const MAX_META = 16 * 1024;
const MAX_RUN_SNAPSHOTS = 8 * 1024 * 1024;
const MAX_SNAPSHOTS = 64 * 1024 * 1024; // serialized bytes; JS heap usage is higher
const MAX_INPUT_BYTES = 64 * 1024;
const MAX_INPUTS = 64;
const ENTRY_BYTES = Symbol('stored bytes');
let snapshotBytes = 0;
let activeVoices = 0;
// Compaction: two adjacent SNAPSHOT entries are considered "nearly
// identical" (newer overwrites older instead of being appended) when
// ≥COMPACT_TEXT_MATCH of their rows have identical text. Markers
// never compact — they're discrete events the viewer needs to see.
const COMPACT_TEXT_MATCH = 0.9;

// Voice input (voice.md). URL of a running whisper-server's /inference
// endpoint. If unset, POST /runs/:id/voice returns 503 — voice is an
// optional add-on; the hub runs fine without it.
const WHISPER_URL = process.env.WHISPER_URL || null;

// Confidence gate thresholds (voice.md) — whisper's stock values; calibrate
// from utterance logs once real usage accumulates.
const GATE_AVG_LOGPROB = -1.0; // below → low_confidence
const GATE_NO_SPEECH = 0.6;    // above (with weak logprob) → no_speech
const GATE_COMPRESSION = 2.4;  // above → low_confidence (repetition loop)

// When set, viewer endpoints require the secret on every connection,
// including loopback. Proxy headers must never decide whether auth applies.
// Unset is for local development only; public hosting requires a secret.
const STREAM_HUB_SECRET = process.env.STREAM_HUB_SECRET || null;

const PUBLIC_DIR = path.join(__dirname, 'public');

const MIME = {
  '.html': 'text/html; charset=utf-8',
  '.js': 'application/javascript; charset=utf-8',
  '.css': 'text/css; charset=utf-8',
  '.json': 'application/json; charset=utf-8',
  '.svg': 'image/svg+xml',
  '.png': 'image/png',
  '.ico': 'image/x-icon',
  '.webmanifest': 'application/manifest+json',
};

/**
 * Per-run state. Every ring entry is a snapshot. voiceInputs are drained
 * alongside inputs on heartbeat but kept separate so the source can prefix
 * voice-origin text with the guide reference (voice.md). lastViewedAt is
 * bumped by the detail-view endpoints (/latest, /history) and relayed to
 * the source as an age so it can pace its heartbeat by viewer presence.
 * touchedAt is the source's report (its own clock) of when the session last
 * drew the user; the viewer orders a machine's sessions by it.
 * @type {Map<string, {
 *   meta: object,
 *   ring: Array<{seq:number, ts:number, payload:any}>,
 *   lastSeen: number,
 *   lastViewedAt: number,
 *   isWorking: boolean,
 *   touchedAt: number|null,
 *   inputs: string[],
 *   voiceInputs: string[],
 * }>}
 */
const runs = new Map();

function removeRun(id) {
  const r = runs.get(id);
  if (r) snapshotBytes -= r.ringBytes;
  runs.delete(id);
}

function dropSnapshot(r) {
  const removed = r.ring.shift();
  r.ringBytes -= removed[ENTRY_BYTES];
  snapshotBytes -= removed[ENTRY_BYTES];
}

function storeSnapshot(r, entry) {
  entry[ENTRY_BYTES] = Buffer.byteLength(JSON.stringify(entry));
  const last = r.ring.at(-1);
  if (last && nearlyIdentical(last.payload, entry.payload)) {
    r.ring.pop();
    r.ringBytes -= last[ENTRY_BYTES];
    snapshotBytes -= last[ENTRY_BYTES];
  }
  r.ring.push(entry);
  r.ringBytes += entry[ENTRY_BYTES];
  snapshotBytes += entry[ENTRY_BYTES];
  while (r.ring.length > RING_CAP || r.ringBytes > MAX_RUN_SNAPSHOTS) dropSnapshot(r);
  while (snapshotBytes > MAX_SNAPSHOTS) {
    let oldest = null;
    for (const candidate of runs.values()) {
      if (candidate.ring.length && (!oldest || candidate.ring[0].ts < oldest.ring[0].ts)) oldest = candidate;
    }
    dropSnapshot(oldest);
  }
}

function queueInput(r, prompt, voice) {
  const bytes = Buffer.byteLength(prompt);
  if (r.inputs.length + r.voiceInputs.length >= MAX_INPUTS || r.inputBytes + bytes > MAX_INPUT_BYTES) return false;
  (voice ? r.voiceInputs : r.inputs).push(prompt);
  r.inputBytes += bytes;
  return true;
}

function validMeta(meta) {
  return meta && !Array.isArray(meta) && typeof meta.runId === 'string' &&
    /^[A-Za-z0-9_-]{1,128}$/.test(meta.runId) &&
    ['cli', 'title', 'host'].every(k => meta[k] === undefined || typeof meta[k] === 'string') &&
    ['startedAt', 'hue'].every(k => meta[k] === undefined || Number.isFinite(meta[k]));
}

function validPayload(payload) {
  return payload && Array.isArray(payload.rows) && payload.rows.length <= 4096 &&
    payload.rows.every(row => row && typeof row.text === 'string' &&
      (row.styles === undefined || (Array.isArray(row.styles) && row.styles.every(style =>
        style && Number.isInteger(style.start) && Number.isInteger(style.end) &&
        style.start >= 0 && style.end >= style.start &&
        ['fg', 'bg'].every(k => style[k] === undefined || typeof style[k] === 'string')))));
}

// Age (not a timestamp): hub and source clocks can't be compared, but
// "how long ago" is clock-free — same reasoning as seq-vs-ts elsewhere.
function viewerAgeMs(r) {
  return r.lastViewedAt ? Date.now() - r.lastViewedAt : null;
}

// Returns true when two snapshot payloads are near-duplicates (same row
// count, ≥COMPACT_TEXT_MATCH rows with identical text). Used to compact
// spinner-frame churn in the ring.
function nearlyIdentical(prev, next) {
  if (!prev || !next) return false;
  if (prev.runId !== next.runId) return false;
  const pr = prev.rows || [];
  const nr = next.rows || [];
  if (pr.length !== nr.length) return false;
  if (pr.length === 0) return true;
  let identical = 0;
  for (let i = 0; i < pr.length; i++) {
    const pt = (pr[i] && pr[i].text) || '';
    const nt = (nr[i] && nr[i].text) || '';
    if (pt === nt) identical++;
  }
  return identical / pr.length >= COMPACT_TEXT_MATCH;
}

function readBodyRaw(req) {
  return new Promise((resolve, reject) => {
    const chunks = [];
    let len = 0;
    req.on('data', (c) => {
      len += c.length;
      if (len > MAX_BODY) {
        req.destroy();
        reject(new Error('payload too large'));
        return;
      }
      chunks.push(c);
    });
    req.on('end', () => resolve(Buffer.concat(chunks)));
    req.on('error', reject);
  });
}

function readBody(req) {
  return readBodyRaw(req).then((b) => b.toString('utf8'));
}

// Decode whatever container the phone's MediaRecorder produced (mp4/AAC on
// iOS Safari, webm/opus elsewhere) into the 16 kHz mono WAV whisper expects.
// Via a temp file, not a pipe: mp4 needs seekable input (moov atom can sit at
// the end). ffmpeg sniffs the container from the bytes; Content-Type is
// ignored.
async function decodeTo16kWav(audio) {
  const dir = await fs.promises.mkdtemp(path.join(os.tmpdir(), 'hub-voice-'));
  const tmp = path.join(dir, 'audio');
  try {
    await fs.promises.writeFile(tmp, audio, { mode: 0o600, flag: 'wx' });
    return await new Promise((resolve, reject) => {
      execFile('ffmpeg', [
        '-hide_banner', '-loglevel', 'error', '-nostdin',
        // Only standalone recording containers; no playlists, network reads,
        // or concat files that can reference arbitrary host files.
        '-protocol_whitelist', 'file',
        '-format_whitelist', 'mov,matroska,webm,wav,ogg',
        '-i', tmp, '-map', '0:a:0', '-vn', '-t', '121',
        '-ar', '16000', '-ac', '1', '-f', 'wav', 'pipe:1',
      ], { encoding: 'buffer', timeout: 15000, killSignal: 'SIGKILL', maxBuffer: MAX_BODY }, (err, wav) => {
        if (err || !wav.length || wav.length > 120 * 16000 * 2 + 1024) {
          reject(new Error('invalid audio or recording exceeds two minutes'));
        } else resolve(wav);
      });
    });
  } finally {
    await fs.promises.rm(dir, { recursive: true, force: true });
  }
}

// One round trip to whisper-server. temperature_inc=1.0 caps the fallback
// ladder at a single retry — garbled audio should fail fast into the phone's
// review path, not burn 3–8s re-decoding (voice.md).
async function transcribe(wav) {
  const form = new FormData();
  form.append('file', new Blob([wav], { type: 'audio/wav' }), 'audio.wav');
  form.append('response_format', 'verbose_json');
  form.append('temperature', '0.0');
  form.append('temperature_inc', '1.0');
  const res = await fetch(WHISPER_URL, { method: 'POST', body: form, signal: AbortSignal.timeout(30000) });
  if (!res.ok) throw new Error(`whisper-server ${res.status}`);
  return res.json();
}

// Utterance-level gate signals: duration-weighted mean log-prob and worst-
// case no-speech from whisper's per-segment stats; compression ratio computed
// here from the transcript (whisper-server's verbose_json omits it, and it's
// just deflate anyway) — high ratio flags repetition-loop hallucinations.
function gateSignals(result, transcript) {
  const segs = result.segments || [];
  let dur = 0;
  let lpSum = 0;
  let noSpeech = 0;
  for (const s of segs) {
    const d = Math.max((s.end ?? 0) - (s.start ?? 0), 0.01);
    dur += d;
    lpSum += (s.avg_logprob ?? 0) * d;
    noSpeech = Math.max(noSpeech, s.no_speech_prob ?? 0);
  }
  const comp = transcript
    ? Buffer.byteLength(transcript, 'utf8') / zlib.deflateSync(Buffer.from(transcript, 'utf8')).length
    : 0;
  return {
    avgLogprob: segs.length ? lpSum / dur : -10,
    noSpeechProb: segs.length ? noSpeech : 1,
    compressionRatio: comp,
  };
}

// Peak 100ms-window RMS of the decoded 16-bit PCM. Zero-energy audio must
// never reach whisper: digital silence is out-of-distribution and decodes to
// confident hallucinated text ("Thank you.") with no_speech_prob ~0 — the
// model signal does NOT fire on it (verified). Server-side VAD would be the
// fuller fix, but whisper-server crashes when VAD filters out all speech.
const SILENCE_RMS = 150; // ~-47 dBFS peak window → treat as no speech
function peakRms(wav) {
  const data = wav.indexOf('data');
  if (data < 0 || data + 8 >= wav.length) return 0;
  const pcm = wav.subarray(data + 8);
  const win = (16000 / 10) * 2; // 100 ms of 16-bit mono samples
  let peak = 0;
  for (let off = 0; off < pcm.length; off += win) {
    const end = Math.min(off + win, pcm.length);
    let sum = 0;
    let n = 0;
    for (let i = off; i + 1 < end; i += 2) {
      const s = pcm.readInt16LE(i);
      sum += s * s;
      n++;
    }
    if (n) peak = Math.max(peak, Math.sqrt(sum / n));
  }
  return peak;
}

// null → confident, auto-send. Whisper hallucinates on silence, so no_speech
// needs its own verdict — a pocket-tap can carry a non-empty transcript.
function gateVerdict(transcript, sig) {
  if (!transcript || (sig.noSpeechProb > GATE_NO_SPEECH && sig.avgLogprob < GATE_AVG_LOGPROB)) return 'no_speech';
  if (sig.avgLogprob < GATE_AVG_LOGPROB || sig.compressionRatio > GATE_COMPRESSION) return 'low_confidence';
  return null;
}

function send(res, code, body, headers = {}) {
  if (typeof body === 'object' && body !== null) {
    body = JSON.stringify(body);
    headers['Content-Type'] = 'application/json; charset=utf-8';
  }
  res.writeHead(code, headers);
  res.end(body);
}

const ok = (res, obj) => send(res, 200, obj);
const noContent = (res) => { res.writeHead(204); res.end(); };
const badRequest = (res, msg) => send(res, 400, (msg || 'bad request') + '\n');
const notFound = (res) => send(res, 404, 'not found\n');

// Serve a static file from public/. Returns true if a file was served.
function serveStatic(req, res, urlPath) {
  if (req.method !== 'GET' && req.method !== 'HEAD') return false;
  if (urlPath.includes('..')) return false;
  const relPath = urlPath === '/' ? '/index.html' : urlPath;
  const filePath = path.join(PUBLIC_DIR, relPath);
  if (!filePath.startsWith(PUBLIC_DIR + path.sep) && filePath !== path.join(PUBLIC_DIR, 'index.html')) return false;
  let stat;
  try { stat = fs.statSync(filePath); } catch { return false; }
  if (!stat.isFile()) return false;
  const ext = path.extname(filePath).toLowerCase();
  const ctype = MIME[ext] || 'application/octet-stream';
  res.writeHead(200, {
    'Content-Type': ctype,
    'Content-Length': stat.size,
    'Cache-Control': 'no-cache',
  });
  if (req.method === 'HEAD') return res.end(), true;
  fs.createReadStream(filePath).pipe(res);
  return true;
}

// Every /runs* request needs the configured secret unless
// explicitly whitelisted.
function requiresAuth(req, path) {
  if (!path.startsWith('/runs')) return false;
  if (isOpenSourcePost(req, path)) return false;
  return true;
}

// Source-side streaming POSTs that don't require the secret. Sources
// (Windows agent-term) can register + push without secrets so corporate
// machines don't need shared-secret distribution.
function isOpenSourcePost(req, path) {
  if (req.method !== 'POST') return false;
  if (path === '/runs') return true;
  if (/^\/runs\/[^/]+\/(snapshot|heartbeat)$/.test(path)) return true;
  return false;
}

async function handle(req, res) {
  res.setHeader('Cache-Control', 'no-store');
  res.setHeader('X-Content-Type-Options', 'nosniff');
  res.setHeader('Referrer-Policy', 'no-referrer');
  res.setHeader('Content-Security-Policy', "default-src 'self'; script-src 'self'; style-src 'self' 'unsafe-inline'; connect-src 'self'; img-src 'self'; media-src 'self' blob:; object-src 'none'; base-uri 'none'; frame-ancestors 'none'; form-action 'self'");
  if (!req.url.startsWith('/') || req.url.startsWith('//')) return badRequest(res, 'invalid path');
  const url = new URL(req.url, 'http://localhost');
  const path = url.pathname;
  // Normal links from other sites may open the viewer; only API requests
  // need the browser-origin checks below.
  if (!path.startsWith('/runs')) {
    if (serveStatic(req, res, path)) return;
    return notFound(res);
  }
  // The viewer is served by this hub. Cross-origin browser clients are not
  // part of the protocol; allowing them exposes even loopback-only hubs.
  const site = req.headers['sec-fetch-site'];
  if (site && site !== 'same-origin' && site !== 'none') return send(res, 403, 'cross-origin request denied\n');
  if (req.headers.origin) {
    let origin;
    try { origin = new URL(req.headers.origin); } catch { return send(res, 403, 'invalid origin\n'); }
    if (!['http:', 'https:'].includes(origin.protocol) || origin.host !== req.headers.host) {
      return send(res, 403, 'cross-origin request denied\n');
    }
  }
  // Host validation also blocks DNS rebinding against unauthenticated local
  // development. A public deployment must set the secret and preserve Host.
  if (!STREAM_HUB_SECRET && (!/^(127\.0\.0\.1|localhost)(:\d+)?$/.test(req.headers.host || '') || req.headers['cf-connecting-ip'])) {
    return send(res, 403, 'configure STREAM_HUB_SECRET for public hosting\n');
  }
  if (req.method === 'OPTIONS') return noContent(res);

  if (STREAM_HUB_SECRET && requiresAuth(req, path)) {
    const supplied = Buffer.from(req.headers['x-hub-secret'] || '');
    const expected = Buffer.from(STREAM_HUB_SECRET);
    if (supplied.length !== expected.length || !crypto.timingSafeEqual(supplied, expected)) {
      return send(res, 401, 'unauthorized\n');
    }
  }

  if (req.method === 'POST' && path.startsWith('/runs') && !path.endsWith('/voice') &&
      (req.headers['content-type'] || '').split(';')[0].trim().toLowerCase() !== 'application/json') {
    return send(res, 415, 'application/json required\n');
  }

  // POST /runs  — register or update meta
  if (req.method === 'POST' && path === '/runs') {
    let meta;
    try { meta = JSON.parse(await readBody(req)); } catch { return badRequest(res, 'invalid json'); }
    if (!validMeta(meta)) return badRequest(res, 'invalid run metadata');
    if (Buffer.byteLength(JSON.stringify(meta)) > MAX_META) return send(res, 413, 'metadata too large\n');
    const now = Date.now();
    const existing = runs.get(meta.runId);
    if (existing) {
      existing.meta = meta;
      existing.lastSeen = now;
    } else {
      if (runs.size >= MAX_RUNS) return send(res, 503, 'run capacity reached\n');
      runs.set(meta.runId, { meta, ring: [], ringBytes: 0, lastSeen: now, lastViewedAt: 0, isWorking: false, touchedAt: null, inputs: [], voiceInputs: [], inputBytes: 0 });
    }
    return noContent(res);
  }

  // GET /runs — list every known run. Viewer renders freshness from lastSeenAt.
  if (req.method === 'GET' && path === '/runs') {
    const list = [];
    for (const r of runs.values()) {
      list.push({ ...r.meta, lastSeenAt: r.lastSeen, isWorking: !!r.isWorking, touchedAt: r.touchedAt });
    }
    return ok(res, { runs: list });
  }

  // /runs/:id/...
  const m = path.match(/^\/runs\/([^\/]+)(?:\/(.+))?$/);
  if (m) {
    let runId;
    try { runId = decodeURIComponent(m[1]); } catch { return badRequest(res, 'invalid runId'); }
    const rest = m[2];

    // DELETE /runs/:id
    if (req.method === 'DELETE' && !rest) {
      removeRun(runId);
      return noContent(res);
    }

    // POST /runs/:id/heartbeat — carries the source's current isWorking flag
    // and touchedAt. Response drains any queued viewer inputs.
    if (req.method === 'POST' && rest === 'heartbeat') {
      const r = runs.get(runId);
      if (!r) return notFound(res);
      let body = {};
      try { body = JSON.parse(await readBody(req)); } catch { return badRequest(res, 'invalid json'); }
      if (!body || typeof body !== 'object' || Array.isArray(body)) return badRequest(res, 'invalid heartbeat');
      if (typeof body.isWorking === 'boolean') r.isWorking = body.isWorking;
      if (Number.isFinite(body.touchedAt)) r.touchedAt = body.touchedAt;
      r.lastSeen = Date.now();
      const inputs = r.inputs;
      r.inputs = [];
      const voiceInputs = r.voiceInputs;
      r.voiceInputs = [];
      r.inputBytes = 0;
      return ok(res, { inputs, voiceInputs, viewerAgeMs: viewerAgeMs(r) });
    }

    // POST /runs/:id/input — viewer-submitted prompt; drained on next heartbeat.
    // source:"voice" marks a reviewed/edited transcript — still a transcript,
    // so the source gives it the guide framing (voice.md).
    if (req.method === 'POST' && rest === 'input') {
      const r = runs.get(runId);
      if (!r) return notFound(res);
      let body;
      try { body = JSON.parse(await readBody(req)); } catch { return badRequest(res, 'invalid json'); }
      if (!body || typeof body.prompt !== 'string' || !body.prompt) return badRequest(res, 'missing prompt');
      if (!queueInput(r, body.prompt, body.source === 'voice')) return send(res, 429, 'input queue full\n');
      return noContent(res);
    }

    // POST /runs/:id/voice — raw audio in (any MediaRecorder container),
    // transcript + gate verdict out; auto-queues on confidence. Contract:
    // voice.md. Synchronous: transcribe + gate + queue in one round trip.
    if (req.method === 'POST' && rest === 'voice') {
      const r = runs.get(runId);
      if (!r) return notFound(res);
      if (!WHISPER_URL) return send(res, 503, 'voice not configured\n');
      if (activeVoices >= 2) return send(res, 429, 'voice busy\n');
      activeVoices++;
      try {
        let audio;
        try { audio = await readBodyRaw(req); } catch { return badRequest(res, 'payload too large'); }
        if (audio.length === 0) return badRequest(res, 'empty body');
        let wav;
        try { wav = await decodeTo16kWav(audio); } catch { return badRequest(res, 'undecodable audio'); }
        if (peakRms(wav) < SILENCE_RMS) {
          return ok(res, { transcript: '', sent: false, holdReason: 'no_speech', signals: null });
        }
        let result;
        try { result = await transcribe(wav); } catch (e) {
          console.error('[voice]', e.message);
          return send(res, 503, 'whisper unavailable\n');
        }
        // Whisper joins segments with newlines; collapse to one line — a
        // newline in typed input would submit the command early at the PTY.
        const transcript = (result.text || '').replace(/\s+/g, ' ').trim();
        const signals = gateSignals(result, transcript);
        const holdReason = gateVerdict(transcript, signals);
        if (!holdReason && !queueInput(r, transcript, true)) return send(res, 429, 'input queue full\n');
        return ok(res, { transcript, sent: !holdReason, holdReason, signals });
      } finally {
        activeVoices--;
      }
    }

    // POST /runs/:id/snapshot — a viewport row array. Compacted when
    // adjacent to a near-identical previous snapshot.
    if (req.method === 'POST' && rest === 'snapshot') {
      const r = runs.get(runId);
      if (!r) return notFound(res);
      let msg;
      try { msg = JSON.parse(await readBody(req)); } catch { return badRequest(res, 'invalid json'); }
      if (!msg || !Number.isSafeInteger(msg.seq) || !validPayload(msg.payload)) return badRequest(res, 'invalid snapshot');
      const stored = { seq: msg.seq, ts: Date.now(), payload: msg.payload };
      storeSnapshot(r, stored);
      r.lastSeen = stored.ts;
      if (msg.payload && typeof msg.payload.isWorking === 'boolean') {
        r.isWorking = msg.payload.isWorking;
      }
      // Snapshot acks carry viewer presence so a WORKING source (30s
      // heartbeats, but frequent snapshots via spinner churn) learns
      // "someone just opened the view" within a second or two instead
      // of a heartbeat period. Old sources ignore the body.
      return ok(res, { viewerAgeMs: viewerAgeMs(r) });
    }

    // GET /runs/:id/latest — viewer's hot-path poll. Returns the most
    // recent snapshot plus a count of viewer-submitted inputs still
    // waiting to be drained by the source on its next heartbeat.
    if (req.method === 'GET' && rest === 'latest') {
      const r = runs.get(runId);
      if (!r) return notFound(res);
      // Detail-view poll = viewer engagement; refreshed every ~1.5s while
      // the view is open. Relayed to the source (heartbeat/snapshot acks)
      // to promote its heartbeat pace while someone is watching.
      r.lastViewedAt = Date.now();
      const snap = r.ring.length > 0 ? r.ring[r.ring.length - 1] : null;
      return ok(res, {
        snapshot: snap,
        lastSeenAt: r.lastSeen,
        isWorking: !!r.isWorking,
        pendingInputs: r.inputs.length + r.voiceInputs.length,
      });
    }

    // GET /runs/:id/history?before=<seq>&limit=N — older snapshots
    // for "load earlier" pagination, oldest→newest.
    if (req.method === 'GET' && rest === 'history') {
      const r = runs.get(runId);
      if (!r) return notFound(res);
      r.lastViewedAt = Date.now();
      const beforeRaw = url.searchParams.get('before');
      const limitRaw = url.searchParams.get('limit');
      const before = beforeRaw == null ? Infinity : Number(beforeRaw);
      const limit = limitRaw == null ? 50 : Math.min(Math.max(1, Number(limitRaw)), RING_CAP);
      const out = [];
      for (let i = r.ring.length - 1; i >= 0 && out.length < limit; i--) {
        if (r.ring[i].seq < before) out.unshift(r.ring[i]);
      }
      return ok(res, { entries: out });
    }
  }

  // Static viewer SPA.
  if (serveStatic(req, res, path)) return;

  return notFound(res);
}

const server = http.createServer((req, res) => {
  handle(req, res).catch((err) => {
    console.error('[err]', err.message);
    if (!res.headersSent) send(res, 500, 'internal\n');
  });
});

setInterval(() => {
  const cutoff = Date.now() - EVICT_MS;
  for (const [id, r] of runs) {
    if (r.lastSeen < cutoff) removeRun(id);
  }
}, 60 * 1000).unref();

server.listen(PORT, HOST, () => {
  console.log(`[hub] listening on http://${HOST}:${server.address().port}`);
});