← back to Cncp Failure Collector

initial scaffold — pm2 + launchd failure collector for CNCP (tick 1: data only, no classifier)

f20012e3448813b407ad2d36238038a165db26e1 · 2026-05-11 10:42:09 -0700 · SteveStudio2

Files touched

• A .gitignore
• A README.md
• A ecosystem.config.js
• A index.js
• A package.json

Diff

commit f20012e3448813b407ad2d36238038a165db26e1
Author: SteveStudio2 <steve@designerwallcoverings.com>
Date:   Mon May 11 10:42:09 2026 -0700

    initial scaffold — pm2 + launchd failure collector for CNCP (tick 1: data only, no classifier)
---
 .gitignore          |   6 ++
 README.md           |  44 +++++++++++
 ecosystem.config.js |  23 ++++++
 index.js            | 206 ++++++++++++++++++++++++++++++++++++++++++++++++++++
 package.json        |  12 +++
 5 files changed, 291 insertions(+)

diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..86ef124
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,6 @@
+node_modules/
+.env*
+tmp/
+*.log
+.DS_Store
+state.json
diff --git a/README.md b/README.md
new file mode 100644
index 0000000..3d4a094
--- /dev/null
+++ b/README.md
@@ -0,0 +1,44 @@
+# cncp-failure-collector
+
+Tails pm2 + launchd for failures and posts them as CNCP tasks. Decided 2026-05-11 via best-practices reviewer + 4-LLM debate team consensus on architecture option B (dedicated `cncp-tasks.json` + tasks panel + collector daemon), tick-1 scope = data pipeline only.
+
+## How it works
+
+Every `POLL_INTERVAL_MS` (default 30 s) the daemon:
+
+1. Runs `pm2 jlist`, diffs against `state.json` (kept in this dir). Any process where `restart_time` bumped or status flipped away from `online` is reported as a failure.
+2. Reads new bytes from `~/Library/Logs/com.steve.*.err.log` files (per-file offsets in `state.json` so a daemon restart doesn't re-post old chunks).
+3. POSTs each event to `http://127.0.0.1:3333/api/failures`. CNCP has a 60-s server-side dedup window per `(source, processName)` to absorb storm bursts.
+
+State file `state.json` is gitignored — it's per-machine runtime data.
+
+## Run
+
+```bash
+pm2 start ecosystem.config.js
+pm2 save
+```
+
+One-shot dev test (no loop, just one scan):
+
+```bash
+npm run once
+```
+
+## Tick 2 (NOT in this tick — scoped out by debate team)
+
+- Hermes 3 (`hermes3:8b`) classifier on **MS1 `http://192.168.1.133:11434`**, NOT localhost. Per-process-name rate limit.
+- POST `/api/failures/:id/classify` returns `{category, severity, suggestedAction}` as async enrichment, not inline.
+- CNCP Failures panel UI (sidebar entry + render card).
+- Optional: outbound webhook to George for high-severity tasks.
+- **launchd error-log tailing** — Steve's convention is per-project paths like `~/Projects/animals/logs/hawk.stderr.log`, NOT `~/Library/Logs/com.steve.*.err.log`. Tick 2 needs to: parse every `~/Library/LaunchAgents/com.steve.*.plist` for `StandardErrorPath`, build a watchlist, tail each. Current code scans `~/Library/Logs/` which is the wrong location for Steve's stack — it no-ops harmlessly today.
+
+## Reviewer flags from tick 1 design (all addressed)
+
+- Local-first: writes only to `127.0.0.1:3333`, no cloud egress.
+- No `ANTHROPIC_API_KEY` read.
+- pm2 poll + per-file tail (not `pm2 logs --json` stream).
+- 60-s server-side dedup window per `(source, processName)`.
+- Daemon survives CNCP being down (retries each cycle).
+- Hardened pm2 entry: max-restarts 10, min-uptime 30 s, exp backoff, 200 M memory cap.
+- Gitified before writing code.
diff --git a/ecosystem.config.js b/ecosystem.config.js
new file mode 100644
index 0000000..23ad2de
--- /dev/null
+++ b/ecosystem.config.js
@@ -0,0 +1,23 @@
+// Hardened for restart storms — autorestart, min-uptime gate, capped restarts,
+// exponential backoff. If 10 rapid restarts happen we stop trying (not into infinity).
+module.exports = {
+  apps: [{
+    name: 'cncp-failure-collector',
+    script: 'index.js',
+    cwd: __dirname,
+    instances: 1,
+    exec_mode: 'fork',
+    autorestart: true,
+    watch: false,
+    max_restarts: 10,
+    min_uptime: '30s',
+    restart_delay: 5000,
+    exp_backoff_restart_delay: 2000,
+    max_memory_restart: '200M',
+    env: {
+      NODE_ENV: 'production',
+      CNCP_URL: 'http://127.0.0.1:3333',
+      POLL_INTERVAL_MS: '30000'
+    }
+  }]
+};
diff --git a/index.js b/index.js
new file mode 100644
index 0000000..7b0eaa0
--- /dev/null
+++ b/index.js
@@ -0,0 +1,206 @@
+#!/usr/bin/env node
+// CNCP failure collector — tick 1 (data only, no Hermes classifier).
+//
+// What it does on each cycle (every POLL_INTERVAL_MS):
+//  1. `pm2 jlist` -> any process that flipped to status != online since last cycle,
+//     or whose restart_time bumped (= crash + auto-restart loop), is reported.
+//  2. Scans ~/.pm2/logs/<name>-error.log + ~/Library/Logs/com.steve.*.err.log
+//     for new bytes since last cycle, posts the new chunk if non-empty.
+//  3. POSTs to CNCP at http://127.0.0.1:3333/api/failures. CNCP dedupes within 60s.
+//
+// State (last restart counts, last error-log byte offsets) lives in state.json
+// in this directory, so a daemon restart doesn't flood CNCP with re-reported events.
+//
+// Best-practices flags addressed:
+//  - pm2 poll, not pm2 logs --json stream (drops lines under volume, schema-volatile)
+//  - per-process dedup (server-side 60s window + here: skip if restart_time unchanged)
+//  - fail-soft: every step in try/catch; one bad cycle doesn't kill the daemon
+//  - no Anthropic SDK; no ANTHROPIC_API_KEY env var read; CNCP is local
+
+const fs = require('fs');
+const path = require('path');
+const os = require('os');
+const { execSync, exec } = require('child_process');
+
+const CNCP_URL = process.env.CNCP_URL || 'http://127.0.0.1:3333';
+const POLL_INTERVAL_MS = parseInt(process.env.POLL_INTERVAL_MS, 10) || 30_000;
+const STATE_FILE = path.join(__dirname, 'state.json');
+const ONCE = process.argv.includes('--once');
+
+const PM2_LOG_DIR = path.join(os.homedir(), '.pm2', 'logs');
+const LAUNCHD_LOG_DIR = path.join(os.homedir(), 'Library', 'Logs');
+
+function log(...a) { console.log(new Date().toISOString(), ...a); }
+
+function loadState() {
+  try {
+    if (fs.existsSync(STATE_FILE)) return JSON.parse(fs.readFileSync(STATE_FILE, 'utf-8'));
+  } catch (e) { log('state.json read failed, starting fresh:', e.message); }
+  return { pm2: {}, logOffsets: {}, startedAt: Date.now() };
+}
+
+function saveState(s) {
+  try { fs.writeFileSync(STATE_FILE, JSON.stringify(s, null, 2), 'utf-8'); }
+  catch (e) { log('state.json write failed:', e.message); }
+}
+
+async function postTask(payload) {
+  try {
+    const r = await fetch(`${CNCP_URL}/api/failures`, {
+      method: 'POST',
+      headers: { 'Content-Type': 'application/json' },
+      body: JSON.stringify(payload)
+    });
+    if (!r.ok) { log('POST /api/failures failed', r.status, await r.text().catch(()=>''));  return null; }
+    return await r.json();
+  } catch (e) {
+    log('POST /api/failures threw:', e.message);
+    return null;
+  }
+}
+
+function readPm2() {
+  try {
+    const out = execSync('pm2 jlist', { encoding: 'utf-8', timeout: 8_000 });
+    return JSON.parse(out);
+  } catch (e) {
+    log('pm2 jlist failed:', e.message);
+    return [];
+  }
+}
+
+function tailFile(p, lines = 20) {
+  try {
+    return execSync(`tail -n ${lines} ${JSON.stringify(p)}`, { encoding: 'utf-8', timeout: 3_000 });
+  } catch { return ''; }
+}
+
+async function scanPm2(state) {
+  const procs = readPm2();
+  const seenNames = new Set();
+  const isBootstrap = Object.keys(state.pm2).length === 0;
+  for (const p of procs) {
+    const name = p.name;
+    seenNames.add(name);
+    const prior = state.pm2[name];
+    const restartTime = p.pm2_env?.restart_time || 0;
+    const status = p.pm2_env?.status || 'unknown';
+    const exitedAt = p.pm2_env?.created_at;
+
+    // On bootstrap (no prior state), seed baseline silently. Don't post historical restarts.
+    if (!prior || isBootstrap) {
+      state.pm2[name] = { restart_time: restartTime, status, exitedAt };
+      continue;
+    }
+
+    const restartedSinceLast = restartTime > prior.restart_time;
+    const flippedDown = prior.status === 'online' && status !== 'online';
+    const persistentlyDown = status !== 'online' && status !== 'launching' && prior.status === status;
+
+    // Report on either a fresh crash-and-restart, OR a fresh flip-to-non-online.
+    // Skip persistently-down (already reported in a prior cycle).
+    if (restartedSinceLast || flippedDown) {
+      const errLog = p.pm2_env?.pm_err_log_path || path.join(PM2_LOG_DIR, `${name}-error.log`);
+      const lastLines = fs.existsSync(errLog) ? tailFile(errLog, 30) : '';
+      const posted = await postTask({
+        source: 'pm2',
+        processName: name,
+        exitCode: p.pm2_env?.exit_code ?? null,
+        lastLines,
+        restartCount: restartTime
+      });
+      if (posted) {
+        log(`reported pm2 failure: ${name} (status=${status}, restarts=${restartTime})`);
+        state.pm2[name] = { restart_time: restartTime, status, exitedAt };
+      } else {
+        log(`pm2 post failed, will retry next cycle: ${name}`);
+        // Don't advance state — next cycle will re-attempt.
+        continue;
+      }
+    } else {
+      state.pm2[name] = { restart_time: restartTime, status, exitedAt };
+    }
+  }
+  // Clean up state for processes that no longer exist (deleted from pm2).
+  for (const k of Object.keys(state.pm2)) {
+    if (!seenNames.has(k)) delete state.pm2[k];
+  }
+}
+
+async function scanLaunchdErrLogs(state) {
+  // Look at ~/Library/Logs/com.steve.*.err.log — only files matching the user's launchd convention.
+  let entries = [];
+  try { entries = fs.readdirSync(LAUNCHD_LOG_DIR); } catch { return; }
+  const errFiles = entries.filter(f => f.startsWith('com.steve.') && /\.err(\.log)?$/.test(f));
+  const isBootstrap = Object.keys(state.logOffsets).length === 0;
+  for (const f of errFiles) {
+    const full = path.join(LAUNCHD_LOG_DIR, f);
+    let stat;
+    try { stat = fs.statSync(full); } catch { continue; }
+    const prior = state.logOffsets[full];
+    const cur = stat.size;
+    // On bootstrap, seed offset at current EOF — don't post pre-existing bytes.
+    if (prior === undefined || isBootstrap) {
+      state.logOffsets[full] = cur;
+      continue;
+    }
+    if (cur <= prior) {
+      // Either no growth, or file was rotated (smaller now). Reset offset to current size.
+      state.logOffsets[full] = cur;
+      continue;
+    }
+    // Read only the new bytes.
+    let chunk = '';
+    try {
+      const fd = fs.openSync(full, 'r');
+      const len = Math.min(cur - prior, 16_000); // cap chunk
+      const buf = Buffer.alloc(len);
+      fs.readSync(fd, buf, 0, len, prior);
+      fs.closeSync(fd);
+      chunk = buf.toString('utf-8');
+    } catch (e) { log('read err log failed', full, e.message); continue; }
+
+    // Suppress trivial noise: blank chunks, lone newlines.
+    const meaningful = chunk.replace(/\s+/g, '').length > 5;
+    if (!meaningful) { state.logOffsets[full] = cur; continue; }
+
+    const processName = f.replace(/^com\.steve\./, '').replace(/\.err(\.log)?$/, '');
+    const posted = await postTask({
+      source: 'launchd',
+      processName,
+      exitCode: null,
+      lastLines: chunk.slice(-4000)
+    });
+    if (posted) {
+      log(`reported launchd error chunk: ${processName} (${chunk.length}B new)`);
+      state.logOffsets[full] = cur;
+    } else {
+      log(`launchd post failed, will retry: ${processName} (offset stays at ${prior})`);
+    }
+  }
+}
+
+async function cycle() {
+  const state = loadState();
+  try { await scanPm2(state); } catch (e) { log('scanPm2 threw:', e.message); }
+  try { await scanLaunchdErrLogs(state); } catch (e) { log('scanLaunchdErrLogs threw:', e.message); }
+  saveState(state);
+}
+
+async function main() {
+  log(`cncp-failure-collector starting. CNCP=${CNCP_URL} POLL=${POLL_INTERVAL_MS}ms ONCE=${ONCE}`);
+  // Health probe: confirm CNCP is reachable before claiming we're up.
+  try {
+    const r = await fetch(`${CNCP_URL}/api/failures?status=triage`);
+    log('CNCP reachable, current open-tasks count:', (await r.json()).length);
+  } catch (e) {
+    log('WARNING: CNCP not reachable at startup — will retry each cycle. err=', e.message);
+  }
+  if (ONCE) { await cycle(); return; }
+  while (true) {
+    try { await cycle(); } catch (e) { log('cycle threw:', e.message); }
+    await new Promise(r => setTimeout(r, POLL_INTERVAL_MS));
+  }
+}
+
+main().catch(e => { console.error('fatal:', e); process.exit(1); });
diff --git a/package.json b/package.json
new file mode 100644
index 0000000..06f91aa
--- /dev/null
+++ b/package.json
@@ -0,0 +1,12 @@
+{
+  "name": "cncp-failure-collector",
+  "version": "0.1.0",
+  "description": "Tails pm2 + launchd for failures and posts them as CNCP tasks. Tick 1: data only. Tick 2: Hermes 3 classifier on MS1.",
+  "main": "index.js",
+  "scripts": {
+    "start": "node index.js",
+    "once": "node index.js --once"
+  },
+  "license": "UNLICENSED",
+  "private": true
+}

(oldest)  ·  back to Cncp Failure Collector  ·  tick 2: heartbeat log every 10 quiet cycles, skip self in pm 495181c →