scripts: track drain-hook-queue.mjs (written 2026-08-05, never committed)

The hook-queue drainer built during the BMF investigation. Its never-delete-on-failure
rule and 10-failure halt are what surfaced the entity-pipeline finding; it was left
untracked in the working tree.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01AB3Kryoy6b1pm2Nz1DYdLh

scripts/drain-hook-queue.mjs

@@ -0,0 +1,174 @@
+#!/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);