Add relay health preflight job and scheduler deferral handling

src/docs/TORCH.md

@@ -13,6 +13,9 @@ AGENT_PLATFORM=codex node bin/torch-lock.mjs lock --agent docs-agent --cadence d
 
 # List all active locks
 node bin/torch-lock.mjs list
+
+# Probe relay health before scheduler lock acquisition
+node bin/torch-lock.mjs health --cadence daily
 ```
 
 ## Defaults
@@ -77,7 +80,7 @@ Operational note: scheduler handoff commands are treated as required execution s
 
 Scheduler failure classes in task logs:
 
-- `backend_unavailable` — lock backend unavailable preflight failures and lock acquisition backend exit code `2` failures/deferrals.
+- `backend_unavailable` — lock backend unavailable preflight failures and lock acquisition backend exit code `2` failures/deferrals; includes relay health alert metadata and incident signal IDs when all relays are unhealthy.
 - `prompt_validation_error` — prompt/handoff, memory evidence, artifact verification, and validation command failures.
 
 

src/lib.mjs

@@ -32,6 +32,7 @@ import {
 } from './services/memory/index.js';
 import { ExitError } from './errors.mjs';
 import { todayDateStr, nowUnix, getIsoWeekStr } from './utils.mjs';
+import { runRelayHealthCheck } from './relay-health.mjs';
 import fs from 'node:fs/promises';
 import path from 'node:path';
 
@@ -527,6 +528,10 @@ function parseArgs(argv) {
     offset: null,
     retentionMs: null,
     windowMs: null,
+    timeoutMs: null,
+    allRelaysDownMinutes: null,
+    minSuccessRate: null,
+    windowMinutes: null,
   };
   let i = 0;
 
@@ -589,6 +594,22 @@ function parseArgs(argv) {
       args.windowMs = parseInt(argv[++i], 10);
     } else if (arg.startsWith('--window-ms=')) {
       args.windowMs = parseInt(arg.split('=')[1], 10);
+    } else if (arg === '--timeout-ms') {
+      args.timeoutMs = parseInt(argv[++i], 10);
+    } else if (arg.startsWith('--timeout-ms=')) {
+      args.timeoutMs = parseInt(arg.split('=')[1], 10);
+    } else if (arg === '--all-relays-down-minutes') {
+      args.allRelaysDownMinutes = parseInt(argv[++i], 10);
+    } else if (arg.startsWith('--all-relays-down-minutes=')) {
+      args.allRelaysDownMinutes = parseInt(arg.split('=')[1], 10);
+    } else if (arg === '--min-success-rate') {
+      args.minSuccessRate = parseFloat(argv[++i]);
+    } else if (arg.startsWith('--min-success-rate=')) {
+      args.minSuccessRate = parseFloat(arg.split('=')[1]);
+    } else if (arg === '--window-minutes') {
+      args.windowMinutes = parseInt(argv[++i], 10);
+    } else if (arg.startsWith('--window-minutes=')) {
+      args.windowMinutes = parseInt(arg.split('=')[1], 10);
     }
   }
 
@@ -603,6 +624,7 @@ Commands:
   lock      --agent <name> --cadence <daily|weekly> Claim a lock
   complete  --agent <name> --cadence <daily|weekly> Mark task as completed (permanent)
   list      [--cadence <daily|weekly>]             Print active lock table
+  health    --cadence <daily|weekly>               Probe relay websocket + publish/read health
   dashboard [--port <port>]                        Serve the dashboard (default: ${DEFAULT_DASHBOARD_PORT})
   init      [--force]                              Initialize torch/ directory in current project
   update    [--force]                              Update torch/ configuration (backups, merges)
@@ -697,6 +719,26 @@ export async function main(argv) {
         break;
       }
 
+      case 'health': {
+        if (!args.cadence || !VALID_CADENCES.has(args.cadence)) {
+          console.error(`ERROR: --cadence <${[...VALID_CADENCES].join('|')}> is required for health`);
+          throw new ExitError(1, 'Missing cadence');
+        }
+        const result = await runRelayHealthCheck({
+          cadence: args.cadence,
+          ...(Number.isFinite(args.timeoutMs) ? { timeoutMs: args.timeoutMs } : {}),
+          ...(Number.isFinite(args.allRelaysDownMinutes) ? { allRelaysDownMinutes: args.allRelaysDownMinutes } : {}),
+          ...(Number.isFinite(args.minSuccessRate) ? { minSuccessRate: args.minSuccessRate } : {}),
+          ...(Number.isFinite(args.windowMinutes) ? { windowMinutes: args.windowMinutes } : {}),
+        });
+        if (!result.ok) {
+          result.failureCategory = 'all relays unhealthy';
+        }
+        console.log(JSON.stringify(result, null, 2));
+        if (!result.ok) throw new ExitError(2, 'Relay health check failed');
+        break;
+      }
+
       case 'dashboard': {
         await cmdDashboard(args.port);
         break;

src/prompts/scheduler-flow.md

@@ -81,16 +81,17 @@ Every agent prompt invoked by the schedulers (daily/weekly) MUST enforce this co
 5. When lock health preflight is enabled (`scheduler.lockHealthPreflight: true` or env `SCHEDULER_LOCK_HEALTH_PREFLIGHT=1`), verify relay/query health before selecting an agent or calling `lock:lock`:
 
    ```bash
-   node scripts/agent/check-relay-health.mjs --cadence <cadence>
+   npm run lock:health -- --cadence <cadence>
    ```
 
    - Escape hatch: set `SCHEDULER_SKIP_LOCK_HEALTH_PREFLIGHT=1` to skip this check for local/offline workflows.
-   - If preflight exits non-zero, write `_failed.md` with reason `Lock backend unavailable preflight` and include:
+   - If preflight exits non-zero because every relay is unhealthy, write `_deferred.md` with reason `All relays unhealthy preflight`, include `incident_signal_id`, and stop before lock acquisition.
+   - For other non-zero preflight failures, write `_failed.md` with reason `Lock backend unavailable preflight` and include:
      - `relay_list`
      - `preflight_failure_category`
      - `preflight_stderr_excerpt`
      - `preflight_stdout_excerpt`
-   - Stop the run immediately when preflight fails.
+   - Always include any preflight alert payload (`preflight_alerts`) in the scheduler metadata for operational triage.
 
 6. Find latest cadence log file, derive the previous agent, then choose the next roster agent not in exclusion set:
 

src/relay-health.mjs

@@ -0,0 +1,281 @@
+import fs from 'node:fs/promises';
+import path from 'node:path';
+import WebSocket from 'ws';
+import { finalizeEvent, generateSecretKey, getPublicKey } from 'nostr-tools/pure';
+import { getNamespace, getRelays } from './torch-config.mjs';
+
+const PROBE_KIND = 27235;
+
+function withTimeout(promise, timeoutMs, message) {
+  let timeoutId;
+  const timeoutPromise = new Promise((_, reject) => {
+    timeoutId = setTimeout(() => reject(new Error(message)), timeoutMs);
+  });
+  return Promise.race([promise, timeoutPromise]).finally(() => clearTimeout(timeoutId));
+}
+
+function nowIso() {
+  return new Date().toISOString();
+}
+
+export function summarizeHistory(historyEntries, { nowMs = Date.now(), windowMinutes = 60 } = {}) {
+  const windowStartMs = nowMs - (windowMinutes * 60_000);
+  const recent = historyEntries.filter((entry) => Date.parse(entry.timestamp || '') >= windowStartMs);
+  const relayProbeCount = recent.reduce((sum, entry) => sum + (entry.summary?.totalRelays || 0), 0);
+  const successCount = recent.reduce((sum, entry) => sum + (entry.summary?.healthyRelays || 0), 0);
+  const successRate = relayProbeCount > 0 ? successCount / relayProbeCount : 1;
+
+  let lastHealthyAtMs = null;
+  for (let i = historyEntries.length - 1; i >= 0; i -= 1) {
+    if ((historyEntries[i].summary?.healthyRelays || 0) > 0) {
+      lastHealthyAtMs = Date.parse(historyEntries[i].timestamp || '');
+      break;
+    }
+  }
+
+  const allDownDurationMinutes = lastHealthyAtMs === null
+    ? null
+    : Math.max(0, Math.floor((nowMs - lastHealthyAtMs) / 60_000));
+
+  return {
+    windowMinutes,
+    sampleCount: recent.length,
+    successRate,
+    allDownDurationMinutes,
+  };
+}
+
+export function evaluateAlertThresholds(currentResult, historyEntries, thresholds, { nowMs = Date.now() } = {}) {
+  const stats = summarizeHistory(historyEntries, { nowMs, windowMinutes: thresholds.windowMinutes });
+  const alerts = [];
+
+  if (currentResult.summary.allRelaysUnhealthy && stats.allDownDurationMinutes !== null && stats.allDownDurationMinutes >= thresholds.allRelaysDownMinutes) {
+    alerts.push({
+      type: 'all_relays_down_duration',
+      thresholdMinutes: thresholds.allRelaysDownMinutes,
+      actualMinutes: stats.allDownDurationMinutes,
+      severity: 'critical',
+    });
+  }
+
+  if (stats.sampleCount > 0 && stats.successRate < thresholds.minSuccessRate) {
+    alerts.push({
+      type: 'relay_success_rate_below_threshold',
+      threshold: thresholds.minSuccessRate,
+      actual: Number(stats.successRate.toFixed(4)),
+      windowMinutes: thresholds.windowMinutes,
+      severity: 'warning',
+    });
+  }
+
+  return { alerts, stats };
+}
+
+async function probeWebSocketReachability(relayUrl, timeoutMs) {
+  const startedAt = Date.now();
+  return withTimeout(new Promise((resolve, reject) => {
+    const ws = new WebSocket(relayUrl);
+    let settled = false;
+
+    const finish = (fn) => (value) => {
+      if (settled) return;
+      settled = true;
+      try {
+        ws.close();
+      } catch {
+        // ignore
+      }
+      fn(value);
+    };
+
+    ws.once('open', finish(() => resolve({ ok: true, latencyMs: Date.now() - startedAt })));
+    ws.once('error', finish((err) => reject(err)));
+  }), timeoutMs, `WebSocket reachability timeout after ${timeoutMs}ms`);
+}
+
+async function probePublishRead(relayUrl, namespace, timeoutMs) {
+  const ws = new WebSocket(relayUrl);
+  const sk = generateSecretKey();
+  const pk = getPublicKey(sk);
+  const createdAt = Math.floor(Date.now() / 1000);
+  const nonce = `${Date.now()}-${Math.random().toString(16).slice(2, 10)}`;
+  const probeTag = `${namespace}-health-probe-${nonce}`;
+  const event = finalizeEvent({
+    kind: PROBE_KIND,
+    created_at: createdAt,
+    tags: [
+      ['t', `${namespace}-relay-health`],
+      ['d', probeTag],
+    ],
+    content: JSON.stringify({ type: 'relay_health_probe', nonce, namespace, pubkey: pk }),
+  }, sk);
+
+  const subId = `probe-${nonce}`;
+
+  return withTimeout(new Promise((resolve, reject) => {
+    let okAck = false;
+    let readSeen = false;
+    let settled = false;
+
+    const settle = (fn) => (value) => {
+      if (settled) return;
+      settled = true;
+      try {
+        ws.close();
+      } catch {
+        // ignore
+      }
+      fn(value);
+    };
+
+    ws.once('open', () => {
+      ws.send(JSON.stringify(['EVENT', event]));
+      ws.send(JSON.stringify(['REQ', subId, { ids: [event.id], limit: 1 }]));
+    });
+
+    ws.on('message', (raw) => {
+      let message;
+      try {
+        message = JSON.parse(raw.toString());
+      } catch {
+        return;
+      }
+      if (!Array.isArray(message) || message.length < 2) return;
+      const type = message[0];
+
+      if (type === 'OK' && message[1] === event.id) {
+        okAck = Boolean(message[2]);
+        if (!okAck) {
+          settle(reject)(new Error(`Relay rejected health probe event: ${String(message[3] || 'unknown reason')}`));
+          return;
+        }
+      }
+
+      if (type === 'EVENT' && message[1] === subId && message[2]?.id === event.id) {
+        readSeen = true;
+      }
+
+      if (okAck && readSeen) {
+        ws.send(JSON.stringify(['CLOSE', subId]));
+        settle(resolve)({ ok: true, eventId: event.id });
+      }
+    });
+
+    ws.once('error', settle(reject));
+  }), timeoutMs, `Publish/read probe timeout after ${timeoutMs}ms`);
+}
+
+async function readHistory(historyPath) {
+  try {
+    const raw = await fs.readFile(historyPath, 'utf8');
+    return raw
+      .split(/\r?\n/)
+      .map((line) => line.trim())
+      .filter(Boolean)
+      .map((line) => {
+        try {
+          return JSON.parse(line);
+        } catch {
+          return null;
+        }
+      })
+      .filter((entry) => entry && typeof entry === 'object');
+  } catch (error) {
+    if (error && error.code === 'ENOENT') return [];
+    throw error;
+  }
+}
+
+async function appendHistory(historyPath, entry) {
+  await fs.mkdir(path.dirname(historyPath), { recursive: true });
+  await fs.appendFile(historyPath, `${JSON.stringify(entry)}\n`, 'utf8');
+}
+
+export async function runRelayHealthCheck(options = {}) {
+  const {
+    cadence = 'daily',
+    relays = getRelays(),
+    namespace = getNamespace(),
+    timeoutMs = 6000,
+    historyPath = path.resolve(process.cwd(), 'task-logs', 'relay-health', `${cadence}.jsonl`),
+    allRelaysDownMinutes = 10,
+    minSuccessRate = 0.7,
+    windowMinutes = 60,
+  } = options;
+
+  const relayResults = [];
+  for (const relay of relays) {
+    const relayResult = {
+      relay,
+      websocketReachable: false,
+      publishReadOk: false,
+      healthy: false,
+      checks: {},
+    };
+
+    try {
+      const wsProbe = await probeWebSocketReachability(relay, timeoutMs);
+      relayResult.websocketReachable = true;
+      relayResult.checks.websocket = wsProbe;
+    } catch (error) {
+      relayResult.checks.websocket = { ok: false, error: error instanceof Error ? error.message : String(error) };
+    }
+
+    if (relayResult.websocketReachable) {
+      try {
+        const prProbe = await probePublishRead(relay, namespace, timeoutMs);
+        relayResult.publishReadOk = true;
+        relayResult.checks.publishRead = prProbe;
+      } catch (error) {
+        relayResult.checks.publishRead = { ok: false, error: error instanceof Error ? error.message : String(error) };
+      }
+    }
+
+    relayResult.healthy = relayResult.websocketReachable && relayResult.publishReadOk;
+    relayResults.push(relayResult);
+  }
+
+  const summary = {
+    totalRelays: relayResults.length,
+    healthyRelays: relayResults.filter((relay) => relay.healthy).length,
+  };
+  summary.unhealthyRelays = summary.totalRelays - summary.healthyRelays;
+  summary.allRelaysUnhealthy = summary.healthyRelays === 0;
+
+  const result = {
+    ok: !summary.allRelaysUnhealthy,
+    cadence,
+    namespace,
+    timestamp: nowIso(),
+    relays,
+    relayResults,
+    summary,
+    incidentSignal: null,
+    alerts: [],
+    alertStats: null,
+    historyPath,
+  };
+
+  const historyEntries = await readHistory(historyPath);
+  const prospectiveHistory = [...historyEntries, result];
+  const thresholdEval = evaluateAlertThresholds(result, prospectiveHistory, {
+    allRelaysDownMinutes,
+    minSuccessRate,
+    windowMinutes,
+  });
+  result.alerts = thresholdEval.alerts;
+  result.alertStats = thresholdEval.stats;
+
+  if (summary.allRelaysUnhealthy) {
+    const incidentId = `relay-health-${cadence}-${Date.now()}`;
+    result.incidentSignal = {
+      id: incidentId,
+      reason: 'All relays unhealthy before lock acquisition',
+      severity: 'critical',
+      suggestedAction: 'Defer scheduler run and investigate relay availability before retrying lock acquisition.',
+    };
+  }
+
+  await appendHistory(historyPath, result);
+  return result;
+}