From 4785d02b7d94afb89d803317b45e3dd926daffe5 Mon Sep 17 00:00:00 2001 From: David F Glidden Date: Thu, 6 Aug 2026 19:45:51 +0200 Subject: [PATCH] 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 Claude-Session: https://claude.ai/code/session_01AB3Kryoy6b1pm2Nz1DYdLh --- scripts/drain-hook-queue.mjs | 174 +++++++++++++++++++++++++++++++++++ 1 file changed, 174 insertions(+) create mode 100644 scripts/drain-hook-queue.mjs diff --git a/scripts/drain-hook-queue.mjs b/scripts/drain-hook-queue.mjs new file mode 100644 index 0000000..2d9f76e --- /dev/null +++ b/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);