dotfiles/scripts/drain-hook-queue.mjs

#!/usr/bin/env node
//
// drain-hook-queue.mjs — drain ~/.capablemind/hook-queue OUT OF BAND.
//
// WHY THIS EXISTS
//   cm-hook.mjs drains at most MAX_QUEUE_DRAIN = 5 queued observations per hook
//   invocation, and the UserPromptSubmit hook is one of those invocations. With a
//   large backlog that means every prompt the steward submits pays for 5 sequential
//   POSTs to BMF — measured 2026-08-06 at ~3.17 s against a 5 s configured timeout
//   (settings.json), which is why "UserPromptSubmit hook timed out after 5s — output
//   discarded" recurs. Draining the backlog here removes the tax permanently; the
//   hook then finds an empty queue and returns fast.
//
// SAFETY
//   * A file is unlinked ONLY after BMF accepts it (res.ok). Never on failure.
//   * On repeated failure the run STOPS and reports what remains. Nothing is dropped.
//   * ENOENT on unlink is tolerated — cm-hook.mjs may drain the same file concurrently
//     (there is no lock; see PENDING-104). Worst case is one duplicate POST, never a loss.
//   * Resumable and idempotent: re-run it, it picks up where it stopped.
//   * --dry-run counts without sending.
//
// DELIBERATE DIVERGENCE FROM cm-hook.mjs
//   The hook treats success as `await res.json() !== null`. A 2xx with an empty or
//   non-JSON body would therefore never be deleted and would stall a bulk run forever.
//   Here success is `res.ok` — the server accepted the observation — and non-JSON
//   bodies are counted and reported separately rather than silently equated.
//
// USAGE
//   node ~/dotfiles/scripts/drain-hook-queue.mjs [--dry-run] [--limit N] [--concurrency N]
//
// Env (same names and defaults as cm-hook.mjs):
//   BM_HOOK_URL (http://localhost:3011) · BM_HOOK_KEY · BM_HOOK_TIMEOUT (ms)

import { readdir, readFile, unlink } from 'node:fs/promises';
import { join } from 'node:path';
import { homedir } from 'node:os';

const BASE_URL = process.env.BM_HOOK_URL || 'http://localhost:3011';
const AUTH_KEY = process.env.BM_HOOK_KEY || 'bm_key_testkey1234567890';
const TIMEOUT = parseInt(process.env.BM_HOOK_TIMEOUT || '10000', 10);
const QUEUE_DIR = join(homedir(), '.capablemind', 'hook-queue');

const argv = process.argv.slice(2);
const flag = (name) => argv.includes(name);
const val = (name, dflt) => {
  const i = argv.indexOf(name);
  return i >= 0 && argv[i + 1] ? parseInt(argv[i + 1], 10) : dflt;
};
const DRY = flag('--dry-run');
const LIMIT = val('--limit', Infinity);
// DEFAULT 1 — MEASURED, NOT ASSUMED (2026-08-06, real queue, real BMF):
//   concurrency  1 → 20/20 deleted in 10.0s  = 2.00/s, 100% success
//   concurrency  4 → 17/20 deleted in 43.0s  = 0.47/s,  85% success
//   concurrency 12 →  3/24 deleted in 20.2s  = 0.15/s,  12% success (21 timeouts)
// Concurrency is ACTIVELY HARMFUL here: /v1/observe is effectively serialized
// server-side, and parallel writers collapse it into timeout queuing. A single
// uncontended POST measured 201 in 1.52s via curl. Raising this number will make
// the drain slower and start failing. Re-measure before changing it.
const CONCURRENCY = Math.max(1, val('--concurrency', 1));
const MAX_CONSECUTIVE_FAILURES = 10;

function log(msg) { process.stdout.write(msg + '\n'); }

async function health() {
  const c = new AbortController();
  const t = setTimeout(() => c.abort(), 5000);
  try {
    const res = await fetch(`${BASE_URL}/health`, { signal: c.signal });
    return res.ok;
  } catch { return false; } finally { clearTimeout(t); }
}

async function post(body) {
  const c = new AbortController();
  const t = setTimeout(() => c.abort(), TIMEOUT);
  try {
    const res = await fetch(`${BASE_URL}/v1/observe`, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json', 'Authorization': `Bearer ${AUTH_KEY}` },
      body: JSON.stringify(body),
      signal: c.signal,
    });
    if (!res.ok) return { ok: false, why: `HTTP ${res.status}` };
    try { await res.json(); return { ok: true, json: true }; }
    catch { return { ok: true, json: false }; }
  } catch (err) {
    return { ok: false, why: err.name === 'AbortError' ? `timeout ${TIMEOUT}ms` : err.message };
  } finally { clearTimeout(t); }
}

const stats = { sent: 0, deleted: 0, nonJson: 0, failed: 0, malformed: 0, vanished: 0 };
let consecutiveFailures = 0;
let stop = false;
let lastWhy = '';

async function drainOne(file) {
  if (stop) return;
  const path = join(QUEUE_DIR, file);
  let body;
  try {
    body = JSON.parse(await readFile(path, 'utf-8'));
  } catch (err) {
    if (err.code === 'ENOENT') { stats.vanished++; return; }   // hook got there first
    stats.malformed++;                                          // unreadable/!JSON — LEAVE IT
    return;
  }
  const r = await post(body);
  stats.sent++;
  if (r.ok) {
    consecutiveFailures = 0;
    if (!r.json) stats.nonJson++;
    try { await unlink(path); stats.deleted++; }
    catch (err) { if (err.code === 'ENOENT') stats.vanished++; else stats.malformed++; }
  } else {
    stats.failed++;
    lastWhy = r.why;
    if (++consecutiveFailures >= MAX_CONSECUTIVE_FAILURES) {
      stop = true;
      log(`\n⚠ STOPPING — ${MAX_CONSECUTIVE_FAILURES} consecutive failures (last: ${r.why}). Nothing deleted on failure.`);
    }
  }
}

const t0 = Date.now();
let files;
try {
  files = (await readdir(QUEUE_DIR)).filter(f => f.endsWith('.json')).sort();
} catch {
  log(`queue dir absent: ${QUEUE_DIR} — nothing to do.`);
  process.exit(0);
}

log(`queue: ${files.length} observation(s) in ${QUEUE_DIR}`);
if (files.length === 0) process.exit(0);

if (DRY) {
  log(`--dry-run: would send ${Math.min(files.length, LIMIT)} to ${BASE_URL}/v1/observe. Nothing sent, nothing deleted.`);
  process.exit(0);
}

// Positive control: never report "drained 0" without establishing BMF is reachable.
if (!(await health())) {
  log(`✗ BMF unreachable at ${BASE_URL}/health — refusing to run. (Nothing was sent or deleted.)`);
  process.exit(1);
}
log(`✓ BMF reachable at ${BASE_URL} — draining with concurrency ${CONCURRENCY}`);

const work = files.slice(0, LIMIT === Infinity ? files.length : LIMIT);
let cursor = 0;
async function worker() {
  while (cursor < work.length && !stop) {
    const i = cursor++;
    await drainOne(work[i]);
    if (stats.sent % 250 === 0 && stats.sent > 0) {
      const rate = stats.sent / ((Date.now() - t0) / 1000);
      log(`  … ${stats.deleted} deleted / ${stats.sent} sent  (${rate.toFixed(1)}/s)`);
    }
  }
}
await Promise.all(Array.from({ length: CONCURRENCY }, worker));

const secs = ((Date.now() - t0) / 1000).toFixed(1);
let remaining = '?';
try { remaining = (await readdir(QUEUE_DIR)).filter(f => f.endsWith('.json')).length; } catch {}

log(`\n=== drain complete in ${secs}s ===`);
log(`  sent:      ${stats.sent}`);
log(`  deleted:   ${stats.deleted}`);
log(`  remaining: ${remaining}`);
if (stats.nonJson)   log(`  accepted with non-JSON body: ${stats.nonJson} (deleted — server accepted)`);
if (stats.vanished)  log(`  vanished mid-flight: ${stats.vanished} (cm-hook drained them concurrently — expected, no lock)`);
if (stats.malformed) log(`  ⚠ malformed/unreadable, LEFT IN PLACE: ${stats.malformed}`);
if (stats.failed)    log(`  ⚠ failed, LEFT IN PLACE: ${stats.failed} (last: ${lastWhy}) — re-run to retry`);
process.exit(stats.failed && stop ? 1 : 0);