Merge S8 (a1): ops — the product watches itself

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

# Conflicts:
#	BUILD-STATE.md
#	CLAUDE.md
#	src/snapshotScheduler.js

src/services/ledgerService.js

@@ -361,6 +361,27 @@ async function defaultGetPlayerStats(name, sport) {
   return { found: false }; // no free settled-result feed yet → pending
 }
 
+/**
+ * Session 8 (A1 board, ops) — count of ledger rows for one game_date (the
+ * daily pulse's "rows written yesterday"). Returns null (NOT 0) when Supabase
+ * isn't configured or the count fails — the pulse renders "n/a", never a
+ * fabricated zero.
+ */
+async function countRowsForDate(gameDate, opts = {}) {
+  if (!gameDate) return null;
+  if (!opts.sb && !isConfigured()) return null;
+  try {
+    const sb = opts.sb || defaultClient();
+    const { count, error } = await sb.from('ledger_entries')
+      .select('id', { count: 'exact', head: true })
+      .eq('game_date', gameDate);
+    if (error) return null;
+    return count || 0;
+  } catch {
+    return null;
+  }
+}
+
 /**
  * 30-day aggregate over the PUBLIC model record (user_id NULL): hit rate,
  * beat-the-close rate, pending count. Percentages are null below
@@ -453,6 +474,7 @@ module.exports = {
   settleLedger,
   settleAllLedgers,
   applyRevision,
+  countRowsForDate,
   getModelAggregate,
   MIN_AGG_SAMPLE,
   __internals: {

src/services/opsWatch.js

@@ -0,0 +1,181 @@
+'use strict';
+
+/**
+ * opsWatch — pure ops-alarm logic for the scheduler (Session 8, A1 board).
+ *
+ * The record must never die silently. This module holds the TESTABLE pieces
+ * the scheduler wires to ntfy (via opsNotify):
+ *
+ *   1. createFailureTracker — per-sport CONSECUTIVE snapshot-failure counter.
+ *      Single-slot errors are normal before books post lines; 3 in a row is a
+ *      broken pipeline. Pages exactly once per losing streak (at the
+ *      threshold crossing), re-arms on any good outcome.
+ *   2. zeroSettleAlarm — "settle pass finished but the record did not
+ *      advance." SIGNAL CHOICE (documented): the ledger settle pass's own
+ *      return values. settleLedger fetches `game_date < today AND outcome IS
+ *      NULL` from POSTGRES, so the signal survives the Redis TTL expiry that
+ *      silently killed morning settles in Session 60 — a
+ *      snapshot:{sport}:previous read would vanish in exactly that failure
+ *      mode, and it costs zero extra reads. Scoped to SETTLEABLE sports (mlb)
+ *      because NBA/WNBA/soccer rows legitimately stay pending until they have
+ *      a settled-result feed. A genuinely empty yesterday fetches 0 rows ->
+ *      settled 0 / pending 0 -> no alarm, ever.
+ *   3. checkQuotaDaily — odds-api >= 80% -> one alert per day (Redis dedupe
+ *      by date key). Complements quotaTracker's once-per-MONTH warn.
+ *   4. buildPulseMessage / countRecentSettles — the 9 AM ET daily pulse, one
+ *      notification, VOICE v1.1: deadpan, numbers, no exclamation points.
+ *
+ * Everything here is pure or fully injectable — no requires of redis/ntfy.
+ */
+
+/** Sports with a real settled-result feed (mlb game logs). Keep in sync with
+ *  outcomeService/ledgerService's MLB-only settlement until Phase 4.5. */
+const SETTLEABLE_SPORTS = ['mlb'];
+
+const PAGE_THRESHOLD = 3;
+
+/** A snapshot result that means "the pipeline produced nothing": a hard error
+ *  or an empty odds feed. 'no grades' (props arrived, grader refused) and 'ok'
+ *  are NOT failures for this counter. */
+function isBadSnapshotResult(result) {
+  if (!result) return false;
+  if (result.status === 'error') return true;
+  return result.status === 'skipped' && String(result.reason || '') === 'no props';
+}
+
+/**
+ * Per-sport consecutive-failure counter (in-process; a restart re-arms, which
+ * is acceptable — the missed-cron watchdog covers a crashed scheduler).
+ * record() returns { sport, count, shouldPage }: shouldPage is true ONLY at
+ * the exact threshold crossing, so a sport that keeps failing pages once per
+ * losing streak, not once per slot.
+ */
+function createFailureTracker(threshold = PAGE_THRESHOLD) {
+  const counts = new Map();
+  return {
+    record(sport, result) {
+      const sp = String(sport || '').toLowerCase();
+      const bad = isBadSnapshotResult(result);
+      const count = bad ? (counts.get(sp) || 0) + 1 : 0;
+      counts.set(sp, count);
+      return { sport: sp, count, shouldPage: bad && count === threshold };
+    },
+    count(sport) { return counts.get(String(sport || '').toLowerCase()) || 0; },
+    threshold,
+  };
+}
+
+/** The daily "close the book" slot: the first configured UTC hour >= 06
+ *  (hours 0-5 UTC are late-night ET slots of the PREVIOUS ET day). Falls back
+ *  to the smallest configured hour when nothing is >= 06. */
+function morningHourUtc(hoursUtc) {
+  const hours = (hoursUtc || []).filter((h) => Number.isInteger(h));
+  if (hours.length === 0) return 14;
+  const daytime = hours.filter((h) => h >= 6);
+  return (daytime.length > 0 ? daytime : hours).sort((a, b) => a - b)[0];
+}
+
+/**
+ * Evaluate the zero-settle signal from settleAllLedgers results.
+ * alarm === true only when settleable-sport rows EXISTED for settlement
+ * (fetched from Postgres: settled + pending > 0) and none settled.
+ */
+function zeroSettleAlarm(ledgerResults, settleable = SETTLEABLE_SPORTS) {
+  const rows = (Array.isArray(ledgerResults) ? ledgerResults : [])
+    .filter((r) => r && settleable.includes(String(r.sport || '').toLowerCase()));
+  const settled = rows.reduce((n, r) => n + (r.settled || 0), 0);
+  const pending = rows.reduce((n, r) => n + (r.pending || 0), 0);
+  return { alarm: settled === 0 && pending > 0, settled, pending };
+}
+
+/** ET calendar date (YYYY-MM-DD) of a Date; UTC fallback if Intl is absent. */
+function dateET(d = new Date()) {
+  try {
+    return new Intl.DateTimeFormat('en-CA', {
+      timeZone: 'America/New_York', year: 'numeric', month: '2-digit', day: '2-digit',
+    }).format(d);
+  } catch {
+    return d.toISOString().slice(0, 10);
+  }
+}
+
+/**
+ * Quota alert: odds-api usage >= thresholdPct -> alert once per UTC day
+ * (Redis dedupe key ops:quota_day:{provider}:{date}). All deps injectable:
+ * { getStatus, cacheGet, cacheSet, notify, now, providerId, thresholdPct }.
+ * Never throws — an ops check must not break the snapshot run.
+ */
+async function checkQuotaDaily(deps = {}) {
+  const providerId = deps.providerId || 'odds-api';
+  const thresholdPct = deps.thresholdPct != null ? deps.thresholdPct : 0.8;
+  try {
+    const status = await deps.getStatus(providerId);
+    const pct = status && Number.isFinite(status.pct) ? status.pct : null;
+    if (pct == null || pct < thresholdPct) return { alerted: false, pct };
+    const day = (deps.now ? deps.now() : new Date()).toISOString().slice(0, 10);
+    const key = `ops:quota_day:${providerId}:${day}`;
+    if (await deps.cacheGet(key)) return { alerted: false, deduped: true, pct };
+    await deps.cacheSet(key, '1', 2 * 24 * 3600);
+    await deps.notify(
+      `${providerId} at ${Math.round(pct * 100)}% of ${status.quotaType || 'period'} quota `
+      + `(${status.used}/${status.limit}). Blocks at 95. PropLine stays primary; `
+      + `the backup path is thinning.`,
+      { title: 'VYNDR quota', priority: 'high', tags: ['warning', 'chart_decreasing'] },
+    );
+    return { alerted: true, pct };
+  } catch (err) {
+    return { alerted: false, error: err.message };
+  }
+}
+
+/** Count settled outcomes across logs whose settledAt falls inside the last
+ *  windowHours (default 24). Pure — feed it the outcomes:{sport}:log arrays. */
+function countRecentSettles(logs, nowMs, windowHours = 24) {
+  const cutoff = nowMs - windowHours * 3600 * 1000;
+  let n = 0;
+  for (const log of logs || []) {
+    for (const o of Array.isArray(log) ? log : []) {
+      const t = o && o.settledAt ? new Date(o.settledAt).getTime() : NaN;
+      if (Number.isFinite(t) && t >= cutoff) n += 1;
+    }
+  }
+  return n;
+}
+
+/**
+ * Assemble the ONE daily-pulse notification body. VOICE v1.1: deadpan,
+ * numbers, no exclamation points. Missing data renders as "n/a" — a pulse
+ * never fabricates a zero.
+ * pieces: { dateEt, ledgerRows (number|null), settles24h (number|null),
+ *           quota ({pct, used, limit}|null), health ({disk_pct, mem_pct}|null) }
+ */
+function buildPulseMessage(pieces = {}) {
+  const na = (v) => (v == null ? 'n/a' : String(v));
+  const quota = pieces.quota && Number.isFinite(pieces.quota.pct)
+    ? `${Math.round(pieces.quota.pct * 100)}% (${pieces.quota.used}/${pieces.quota.limit})`
+    : 'n/a';
+  const h = pieces.health || {};
+  const disk = Number.isFinite(h.disk_pct) ? `${h.disk_pct}%` : 'n/a';
+  const mem = Number.isFinite(h.mem_pct) ? `${h.mem_pct}%` : 'n/a';
+  return [
+    `VYNDR pulse — ${pieces.dateEt || dateET()}`,
+    `ledger rows yesterday: ${na(pieces.ledgerRows)}`,
+    `settles last 24h: ${na(pieces.settles24h)}`,
+    `odds-api quota: ${quota}`,
+    `disk ${disk} · mem ${mem}`,
+    'desk pack: see /desk',
+  ].join('\n');
+}
+
+module.exports = {
+  createFailureTracker,
+  isBadSnapshotResult,
+  zeroSettleAlarm,
+  morningHourUtc,
+  checkQuotaDaily,
+  countRecentSettles,
+  buildPulseMessage,
+  dateET,
+  SETTLEABLE_SPORTS,
+  PAGE_THRESHOLD,
+};

src/services/systemHealth.js

@@ -0,0 +1,60 @@
+'use strict';
+
+/**
+ * systemHealth — box vitals (Session 8, A1 board — ops).
+ *
+ * Pure + injectable: disk usage via fs.promises.statfs('/'), memory via
+ * os.freemem/os.totalmem. Returns percentages (0-100, rounded) or null when a
+ * probe fails — the caller renders "n/a", never a fabricated number (Data
+ * Semantics Rule applies to ops copy too). Zero dependencies, zero cost.
+ */
+
+const fsp = require('fs').promises;
+const os = require('os');
+
+/** Alerting thresholds — checked by the daily pulse. */
+const THRESHOLDS = { disk_pct: 85, mem_pct: 90 };
+
+const roundPct = (frac) => Math.min(100, Math.max(0, Math.round(frac * 100)));
+
+/**
+ * { disk_pct, mem_pct } — each null when its probe fails.
+ * opts: { fsImpl (statfs), osImpl (freemem/totalmem), path }.
+ */
+async function getSystemHealth(opts = {}) {
+  const fsImpl = opts.fsImpl || fsp;
+  const osImpl = opts.osImpl || os;
+  const out = { disk_pct: null, mem_pct: null };
+  try {
+    const s = await fsImpl.statfs(opts.path || '/');
+    if (s && Number.isFinite(s.blocks) && s.blocks > 0 && Number.isFinite(s.bavail)) {
+      out.disk_pct = roundPct(1 - s.bavail / s.blocks);
+    }
+  } catch { /* disk_pct stays null */ }
+  try {
+    const total = osImpl.totalmem();
+    const free = osImpl.freemem();
+    if (Number.isFinite(total) && total > 0 && Number.isFinite(free)) {
+      out.mem_pct = roundPct(1 - free / total);
+    }
+  } catch { /* mem_pct stays null */ }
+  return out;
+}
+
+/**
+ * Deadpan issue lines for anything over threshold (strictly >). Empty array =
+ * healthy. Copy carries the numbers; no punctuation theatrics.
+ */
+function healthIssues(health, thresholds = THRESHOLDS) {
+  const issues = [];
+  if (!health) return issues;
+  if (Number.isFinite(health.disk_pct) && health.disk_pct > thresholds.disk_pct) {
+    issues.push(`disk at ${health.disk_pct}% (threshold ${thresholds.disk_pct}%)`);
+  }
+  if (Number.isFinite(health.mem_pct) && health.mem_pct > thresholds.mem_pct) {
+    issues.push(`memory at ${health.mem_pct}% (threshold ${thresholds.mem_pct}%)`);
+  }
+  return issues;
+}
+
+module.exports = { getSystemHealth, healthIssues, THRESHOLDS };

src/snapshotScheduler.js

@@ -60,9 +60,19 @@ function startSnapshotScheduler(opts = {}) {
   const settleLedgers = opts.settleAllLedgers || require('./services/ledgerService').settleAllLedgers;
   const notify = opts.notify || require('./utils/opsNotify').notify;
   const cacheGet = opts.cacheGet || require('./utils/redis').cacheGet;
+  const cacheSet = opts.cacheSet || require('./utils/redis').cacheSet;
   const now = opts.now || (() => new Date());
+  // Session 8 (A1 board) — ops: the product watches itself. Pure logic lives
+  // in services/opsWatch + services/systemHealth; this file only wires it.
+  const opsWatch = require('./services/opsWatch');
+  const getSystemHealth = opts.getSystemHealth || require('./services/systemHealth').getSystemHealth;
+  const healthIssues = opts.healthIssues || require('./services/systemHealth').healthIssues;
+  const getQuotaStatus = opts.getQuotaStatus || require('./services/quotaTracker').getQuotaStatus;
+  const countLedgerRows = opts.countLedgerRows || require('./services/ledgerService').countRowsForDate;
+  const failureTracker = opts.failureTracker || opsWatch.createFailureTracker();
   let lastFiredSlot = null;
   let lastOverdueSlot = null;
+  let lastPulseDate = null;
 
   // Session 56 — missed-cron watchdog. Runs every minute (independent of the
   // fire schedule): if a scheduled slot came and went without a snapshot, alert
@@ -83,8 +93,55 @@ function startSnapshotScheduler(opts = {}) {
     } catch { /* watchdog must never throw */ }
   };
 
+  // Session 8 — daily 9 AM ET pulse (13:00 UTC), ONE notification. Dedupe is
+  // belt-and-braces: in-process date + a Redis key that survives a restart
+  // inside the same day. Runs on the per-minute cadence, independent of slots.
+  const PULSE_HOUR_UTC = Number.parseInt(process.env.PULSE_HOUR_UTC || '13', 10);
+  const pulseTick = async () => {
+    try {
+      const d = now();
+      if (d.getUTCHours() !== PULSE_HOUR_UTC || d.getUTCMinutes() !== 0) return;
+      const dayKey = d.toISOString().slice(0, 10);
+      if (dayKey === lastPulseDate) return;
+      lastPulseDate = dayKey;
+      if (await cacheGet(`ops:pulse:${dayKey}`)) return;   // already sent today (pre-restart)
+      await cacheSet(`ops:pulse:${dayKey}`, '1', 2 * 24 * 3600);
+
+      const yesterdayEt = opsWatch.dateET(new Date(d.getTime() - 24 * 3600 * 1000));
+      let ledgerRows = null;
+      try { ledgerRows = await countLedgerRows(yesterdayEt); } catch { /* n/a */ }
+      let settles24h = null;
+      try {
+        const logs = [];
+        for (const sp of ['mlb', 'nba', 'wnba', 'soccer']) {
+          const raw = await cacheGet(`outcomes:${sp}:log`);
+          logs.push(Array.isArray(raw) ? raw : (raw && raw.log) || []);
+        }
+        settles24h = opsWatch.countRecentSettles(logs, d.getTime());
+      } catch { /* n/a */ }
+      let quota = null;
+      try { quota = await getQuotaStatus('odds-api'); } catch { /* n/a */ }
+      let health = null;
+      try { health = await getSystemHealth(); } catch { /* n/a */ }
+
+      await notify(
+        opsWatch.buildPulseMessage({ dateEt: opsWatch.dateET(d), ledgerRows, settles24h, quota, health }),
+        { title: 'VYNDR daily pulse', tags: ['newspaper'] },
+      );
+      // Box health pages separately at high priority — the pulse is a read,
+      // the threshold breach is an action item.
+      const issues = healthIssues(health);
+      if (issues.length > 0) {
+        await notify(`Box health: ${issues.join('; ')}. The record does not fit on a full disk.`, {
+          title: 'VYNDR box', priority: 'high', tags: ['warning'],
+        });
+      }
+    } catch { /* the pulse must never break the scheduler */ }
+  };
+
   const tick = async () => {
     await checkOverdue();
+    await pulseTick();
     const d = now();
     if (d.getUTCMinutes() !== 0) return;
     const h = d.getUTCHours();
@@ -92,20 +149,47 @@ function startSnapshotScheduler(opts = {}) {
     const slot = `${d.getUTCFullYear()}-${d.getUTCMonth()}-${d.getUTCDate()}-${h}`;
     if (slot === lastFiredSlot) return; // fire once per slot
     lastFiredSlot = slot;
+    // Session 8 — a settle pass that THROWS is a paged event, not a log line.
+    // Settlement failing quietly is how the record dies.
     try {
       const settled = await settleAll();
       const totalSettled = settled.reduce((n, r) => n + (r.settled || 0), 0);
       console.log(`[outcomes] cron fired ${h}:00 UTC — ${totalSettled} props settled vs real results`);
     } catch (e) {
       console.warn('[outcomes] settle run failed:', e.message);
+      await notify(`Outcome settlement THREW at ${h}:00 UTC — ${e.message}. Yesterday's grades are unsettled until the next slot.`, {
+        title: 'VYNDR settlement', priority: 'high', tags: ['rotating_light'],
+      });
     }
+    let ledgerResults = null;
     try {
-      const ledger = await settleLedgers();
-      const n = ledger.reduce((t, r) => t + (r.settled || 0), 0);
+      ledgerResults = await settleLedgers();
+      const n = ledgerResults.reduce((t, r) => t + (r.settled || 0), 0);
       console.log(`[ledger] settle pass — ${n} entries settled (outcome + CLV)`);
     } catch (e) {
       console.warn('[ledger] settle run failed:', e.message);
+      await notify(`Ledger settlement THREW at ${h}:00 UTC — ${e.message}. The public record did not advance.`, {
+        title: 'VYNDR settlement', priority: 'high', tags: ['rotating_light'],
+      });
     }
+    // Session 8 — zero-settle alarm, MORNING slot only (the book-closing pass).
+    // Signal = the ledger settle's own Postgres-backed return values (see
+    // opsWatch.zeroSettleAlarm for why not snapshot:{sport}:previous). Deduped
+    // once per ET date; a genuinely empty yesterday (0 rows fetched) never fires.
+    try {
+      if (h === opsWatch.morningHourUtc(HOURS_UTC) && ledgerResults) {
+        const z = opsWatch.zeroSettleAlarm(ledgerResults);
+        if (z.alarm) {
+          const dk = `ops:settle_zero:${opsWatch.dateET(d)}`;
+          if (!(await cacheGet(dk))) {
+            await cacheSet(dk, '1', 2 * 24 * 3600);
+            await notify(`Morning settle pass closed with 0 settles — ${z.pending} ledger rows from before today are still pending. Yesterday had graded rows; the record did not advance.`, {
+              title: 'VYNDR settlement', priority: 'high', tags: ['rotating_light'],
+            });
+          }
+        }
+      }
+    } catch { /* alarm evaluation must never break the tick */ }
     try {
       const results = await runAll();
       const ok = results.filter((r) => r.status === 'ok');
@@ -116,9 +200,22 @@ function startSnapshotScheduler(opts = {}) {
         const total = results.reduce((n, r) => n + (r.gradeCount || 0), 0);
         await notify(`Desk pack ready — ${total} props graded across ${ok.length} sports. vyndr.app/desk`, { title: 'VYNDR desk', tags: ['newspaper'] });
       }
+      // Session 8 — persistent-failure pager: 3+ CONSECUTIVE erroring slots for
+      // a sport pages once (single-slot errors are normal before lines post).
+      for (const r of results) {
+        const t = failureTracker.record(r.sport, r);
+        if (t.shouldPage) {
+          await notify(`${String(r.sport).toUpperCase()} snapshot has produced nothing for ${t.count} consecutive slots (latest: ${r.status}${r.reason ? ` — ${r.reason}` : ''}). One empty slot is a quiet book; ${t.count} is a broken pipe.`, {
+            title: 'VYNDR pipeline', priority: 'high', tags: ['rotating_light'],
+          });
+        }
+      }
     } catch (e) {
       console.warn('[snapshot] cron run failed:', e.message);
     }
+    // Session 8 — quota check after each snapshot run: odds-api >= 80% alerts
+    // once per day (Redis-deduped). Never throws (guarded inside checkQuotaDaily).
+    await opsWatch.checkQuotaDaily({ getStatus: getQuotaStatus, cacheGet, cacheSet, notify, now: () => now() });
   };
 
   // Session 60 (night2/D) — Phase 2.5 intraday refresh. Every
@@ -158,7 +255,9 @@ function startSnapshotScheduler(opts = {}) {
   // was invisible at boot, which made "is settlement scheduled?" unanswerable
   // from logs. This line makes it verifiable forever.
   console.log(`[settlement] armed — outcomes + ledger settle pass runs FIRST at each snapshot slot (${HOURS_UTC.join(',')} UTC), idempotent re-runs`);
-  return { interval, tick, refreshTick };
+  // Session 8 — same verifiability rule: every watchdog states itself at boot.
+  console.log(`[opsWatch] armed — settle alarms (throw + morning zero-settle), failure pager (${failureTracker.threshold} consecutive), quota daily check, pulse ${PULSE_HOUR_UTC}:00 UTC`);
+  return { interval, tick, refreshTick, pulseTick };
 }
 
 module.exports = { startSnapshotScheduler, HOURS_UTC, mostRecentExpectedSlot, isSnapshotOverdue };