Merge branch 'main' into refactor-cmdinit-ops-5287196156995262612

src/cmd-check.mjs

@@ -8,10 +8,10 @@ import { queryLocks } from './lock-ops.mjs';
 import { todayDateStr } from './utils.mjs';
 
 export async function cmdCheck(cadence) {
-  const relays = getRelays();
-  const namespace = getNamespace();
+  const relays = await getRelays();
+  const namespace = await getNamespace();
   const dateStr = todayDateStr();
-  const config = loadTorchConfig();
+  const config = await loadTorchConfig();
   const pausedAgents = (cadence === 'daily' ? config.scheduler.paused.daily : config.scheduler.paused.weekly) || [];
 
   console.error(`Checking locks: namespace=${namespace}, cadence=${cadence}, date=${dateStr}`);
@@ -20,9 +20,9 @@ export async function cmdCheck(cadence) {
     console.error(`Paused agents: ${pausedAgents.join(', ')}`);
   }
 
-  const locks = await queryLocks(relays, cadence, dateStr, namespace);
+  const locks = await queryLocksFn(relays, cadence, dateStr, namespace);
   const lockedAgents = [...new Set(locks.map((l) => l.agent).filter(Boolean))];
-  const roster = getRoster(cadence);
+  const roster = await getRoster(cadence);
 
   const excludedAgents = [...new Set([...lockedAgents, ...pausedAgents])];
   const unknownLockedAgents = lockedAgents.filter((agent) => !roster.includes(agent));

src/cmd-list.mjs

@@ -7,8 +7,8 @@ import { queryLocks } from './lock-ops.mjs';
 import { todayDateStr, nowUnix } from './utils.mjs';
 
 export async function cmdList(cadence) {
-  const relays = getRelays();
-  const namespace = getNamespace();
+  const relays = await getRelays();
+  const namespace = await getNamespace();
   const dateStr = todayDateStr();
   const cadences = cadence ? [cadence] : ['daily', 'weekly'];
 
@@ -47,7 +47,7 @@ export async function cmdList(cadence) {
       );
     }
 
-    const roster = getRoster(c);
+    const roster = await getRoster(c);
     const lockedAgents = new Set(locks.map((l) => l.agent).filter(Boolean));
     const unknownLockedAgents = [...lockedAgents].filter((agent) => !roster.includes(agent));
     const available = roster.filter((a) => !lockedAgents.has(a));

src/cmd-lock.mjs

@@ -14,10 +14,10 @@ import { todayDateStr, nowUnix } from './utils.mjs';
 import { ExitError } from './errors.mjs';
 
 export async function cmdLock(agent, cadence, dryRun = false) {
-  const relays = getRelays();
-  const namespace = getNamespace();
+  const relays = await getRelays();
+  const namespace = await getNamespace();
   const dateStr = todayDateStr();
-  const ttl = getTtl();
+  const ttl = await getTtl();
   const now = nowUnix();
   const expiresAt = now + ttl;
 
@@ -25,7 +25,7 @@ export async function cmdLock(agent, cadence, dryRun = false) {
   console.error(`TTL: ${ttl}s, expires: ${new Date(expiresAt * 1000).toISOString()}`);
   console.error(`Relays: ${relays.join(', ')}`);
 
-  const roster = getRoster(cadence);
+  const roster = await getRoster(cadence);
   if (!roster.includes(agent)) {
     console.error(`ERROR: agent "${agent}" is not in the ${cadence} roster`);
     console.error(`Allowed ${cadence} agents: ${roster.join(', ')}`);

src/dashboard.mjs

@@ -22,9 +22,10 @@ export async function cmdDashboard(port = DEFAULT_DASHBOARD_PORT, host = '127.0.
   // Resolve package root relative to this file (src/dashboard.mjs)
   // this file is in <root>/src/dashboard.mjs, so '..' goes to <root>
   const packageRoot = path.resolve(path.dirname(fileURLToPath(import.meta.url)), '..');
-  const auth = getDashboardAuth();
+  const auth = await getDashboardAuth();
 
   const server = http.createServer(async (req, res) => {
+    try {
     // Basic Auth check
     if (auth) {
       const authHeader = req.headers.authorization;
@@ -181,20 +182,28 @@ export async function cmdDashboard(port = DEFAULT_DASHBOARD_PORT, host = '127.0.
 
     res.writeHead(200, { 'Content-Type': contentType });
     fs.createReadStream(filePath).pipe(res);
-  });
-
-  server.listen(port, host, () => {
-    const listenUrl = `http://${host === '0.0.0.0' ? 'localhost' : host}:${port}/dashboard/`;
-    console.log(`Dashboard running at ${listenUrl}`);
-    if (auth) {
-      console.log('Authentication: enabled (Basic Auth)');
-    } else {
-      console.warn('Authentication: DISABLED (Dashboard is public)');
+    } catch (err) {
+      console.error('Dashboard Server Error:', err);
+      if (!res.headersSent) {
+        res.writeHead(500);
+        res.end('Internal Server Error');
+      }
     }
-    console.log(`Serving files from ${packageRoot}`);
-    console.log(`Using configuration from ${process.cwd()}`);
   });
 
-  // Keep process alive
-  return server;
+  return new Promise((resolve, reject) => {
+    server.listen(port, host, () => {
+      const listenUrl = `http://${host === '0.0.0.0' ? 'localhost' : host}:${port}/dashboard/`;
+      console.log(`Dashboard running at ${listenUrl}`);
+      if (auth) {
+        console.log('Authentication: enabled (Basic Auth)');
+      } else {
+        console.warn('Authentication: DISABLED (Dashboard is public)');
+      }
+      console.log(`Serving files from ${packageRoot}`);
+      console.log(`Using configuration from ${process.cwd()}`);
+      resolve(server);
+    });
+    server.on('error', reject);
+  });
 }

src/lib.mjs

@@ -75,10 +75,10 @@ export async function cmdCheck(cadence, deps = {}) {
     quiet = false,
   } = deps;
 
-  const relays = getRelays();
-  const namespace = getNamespace();
+  const relays = await getRelays();
+  const namespace = await getNamespace();
   const dateStr = getDateStr();
-  const config = loadTorchConfig();
+  const config = await loadTorchConfig();
   const pausedAgents = config.scheduler.paused[cadence] || [];
 
   if (!quiet) {
@@ -99,7 +99,7 @@ export async function cmdCheck(cadence, deps = {}) {
 
   const locks = await queryLocks(relays, cadence, dateStr, namespace);
   const lockedAgents = [...new Set(locks.map((l) => l.agent).filter(Boolean))];
-  const roster = getRoster(cadence);
+  const roster = await getRoster(cadence);
   const rosterSet = new Set(roster);
 
   const excludedAgentsSet = new Set([...lockedAgents, ...pausedAgents, ...completedAgents]);
@@ -180,11 +180,11 @@ export async function cmdLock(agent, cadence, optionsOrDryRun = false, deps = {}
     error = console.error
   } = deps;
 
-  const relays = getRelays();
-  const namespace = getNamespace();
-  const hashtag = getHashtag();
+  const relays = await getRelays();
+  const namespace = await getNamespace();
+  const hashtag = await getHashtag();
   const dateStr = getDateStr();
-  const ttl = getTtl();
+  const ttl = await getTtl();
   const now = nowUnix();
   const expiresAt = now + ttl;
 
@@ -193,7 +193,7 @@ export async function cmdLock(agent, cadence, optionsOrDryRun = false, deps = {}
   error(`TTL: ${ttl}s, expires: ${new Date(expiresAt * 1000).toISOString()}`);
   error(`Relays: ${relays.join(', ')}`);
 
-  const roster = getRoster(cadence);
+  const roster = await getRoster(cadence);
   if (!roster.includes(agent)) {
     error(`ERROR: agent "${agent}" is not in the ${cadence} roster`);
     error(`Allowed ${cadence} agents: ${roster.join(', ')}`);
@@ -319,8 +319,8 @@ export async function cmdList(cadence, deps = {}) {
     error = console.error
   } = deps;
 
-  const relays = getRelays();
-  const namespace = getNamespace();
+  const relays = await getRelays();
+  const namespace = await getNamespace();
   const dateStr = getDateStr();
   const cadences = cadence ? [cadence] : [...VALID_CADENCES];
 
@@ -365,7 +365,7 @@ export async function cmdList(cadence, deps = {}) {
       );
     }
 
-    const roster = getRoster(c);
+    const roster = await getRoster(c);
     const rosterSet = new Set(roster);
     const lockedAgents = new Set(locks.map((l) => l.agent).filter(Boolean));
     const unknownLockedAgents = [...lockedAgents].filter((agent) => !rosterSet.has(agent));
@@ -413,9 +413,9 @@ export async function cmdComplete(agent, cadence, optionsOrDryRun = false, deps
     error = console.error
   } = deps;
 
-  const relays = getRelays();
-  const namespace = getNamespace();
-  const hashtag = getHashtag();
+  const relays = await getRelays();
+  const namespace = await getNamespace();
+  const hashtag = await getHashtag();
   const dateStr = getDateStr();
   const now = nowUnix();
 

src/lock-ops.mjs

@@ -8,11 +8,6 @@ import {
 } from './torch-config.mjs';
 import { KIND_APP_DATA } from './constants.mjs';
 
-const relayHealthState = {
-  metricsByRelay: new Map(),
-  lastSnapshotAt: 0,
-};
-
 function nowUnix() {
   return Math.floor(Date.now() / 1000);
 }
@@ -139,23 +134,6 @@ function calculateBackoffDelayMs(attemptNumber, baseMs, capMs, randomFn = Math.r
   return Math.floor(randomFn() * maxDelay);
 }
 
-function ensureRelayMetrics(relay, config) {
-  let metrics = relayHealthState.metricsByRelay.get(relay);
-  if (!metrics) {
-    metrics = {
-      relay,
-      recentOutcomes: [],
-      failureStreak: 0,
-      quarantineUntil: 0,
-      cooldownMs: config.quarantineCooldownMs,
-      lastLatencyMs: null,
-      lastResultAt: null,
-    };
-    relayHealthState.metricsByRelay.set(relay, metrics);
-  }
-  return metrics;
-}
-
 function summarizeRelayMetrics(metrics, nowMs) {
   const total = metrics.recentOutcomes.length;
   const successCount = metrics.recentOutcomes.filter((entry) => entry.success).length;
@@ -189,95 +167,126 @@ function computeRelayScore(summary) {
   return (summary.successRate * 1.4) - (summary.timeoutRate * 0.9) - latencyPenalty - quarantinePenalty;
 }
 
-function rankRelaysByHealth(relays, config, nowMs = Date.now()) {
-  const summaries = relays.map((relay) => {
-    const metrics = ensureRelayMetrics(relay, config);
-    const summary = summarizeRelayMetrics(metrics, nowMs);
+export class RelayHealthManager {
+  constructor() {
+    this.metricsByRelay = new Map();
+    this.lastSnapshotAt = 0;
+  }
+
+  ensureMetrics(relay, config) {
+    let metrics = this.metricsByRelay.get(relay);
+    if (!metrics) {
+      metrics = {
+        relay,
+        recentOutcomes: [],
+        failureStreak: 0,
+        quarantineUntil: 0,
+        cooldownMs: config.quarantineCooldownMs,
+        lastLatencyMs: null,
+        lastResultAt: null,
+      };
+      this.metricsByRelay.set(relay, metrics);
+    }
+    return metrics;
+  }
+
+  rankRelays(relays, config, nowMs = Date.now()) {
+    const summaries = relays.map((relay) => {
+      const metrics = this.ensureMetrics(relay, config);
+      const summary = summarizeRelayMetrics(metrics, nowMs);
+      return {
+        relay,
+        summary,
+        score: computeRelayScore(summary),
+      };
+    });
+
+    return summaries.sort((a, b) => {
+      if (b.score !== a.score) return b.score - a.score;
+      if (a.summary.quarantined !== b.summary.quarantined) return a.summary.quarantined ? 1 : -1;
+      if (a.summary.averageLatencyMs !== b.summary.averageLatencyMs) {
+        if (a.summary.averageLatencyMs === null) return 1;
+        if (b.summary.averageLatencyMs === null) return -1;
+        return a.summary.averageLatencyMs - b.summary.averageLatencyMs;
+      }
+      return a.relay.localeCompare(b.relay);
+    });
+  }
+
+  prioritizeRelays(relays, config, nowMs = Date.now()) {
+    const minActive = Math.max(1, Math.min(config.minActiveRelayPool, relays.length || 1));
+    const ranked = this.rankRelays([...new Set(relays)], config, nowMs);
+    const active = ranked.filter((entry) => !entry.summary.quarantined);
+    const quarantined = ranked
+      .filter((entry) => entry.summary.quarantined)
+      .sort((a, b) => a.summary.quarantineRemainingMs - b.summary.quarantineRemainingMs);
+
+    const selected = [...active];
+    if (selected.length < minActive) {
+      const additionalNeeded = Math.min(minActive - selected.length, quarantined.length);
+      selected.push(...quarantined.slice(0, additionalNeeded));
+    }
+
     return {
-      relay,
-      summary,
-      score: computeRelayScore(summary),
+      prioritized: selected.map((entry) => entry.relay),
+      ranked,
     };
-  });
+  }
 
-  return summaries.sort((a, b) => {
-    if (b.score !== a.score) return b.score - a.score;
-    if (a.summary.quarantined !== b.summary.quarantined) return a.summary.quarantined ? 1 : -1;
-    if (a.summary.averageLatencyMs !== b.summary.averageLatencyMs) {
-      if (a.summary.averageLatencyMs === null) return 1;
-      if (b.summary.averageLatencyMs === null) return -1;
-      return a.summary.averageLatencyMs - b.summary.averageLatencyMs;
+  recordOutcome(relay, success, errorMessage, latencyMs, config, nowMs = Date.now()) {
+    const metrics = this.ensureMetrics(relay, config);
+    const timedOut = String(errorMessage || '').toLowerCase().includes('timeout')
+      || String(errorMessage || '').toLowerCase().includes('timed out');
+    metrics.recentOutcomes.push({ success, timedOut, latencyMs, atMs: nowMs });
+    if (metrics.recentOutcomes.length > config.rollingWindowSize) {
+      metrics.recentOutcomes.splice(0, metrics.recentOutcomes.length - config.rollingWindowSize);
     }
-    return a.relay.localeCompare(b.relay);
-  });
-}
+    metrics.lastResultAt = nowMs;
+    metrics.lastLatencyMs = Number.isFinite(latencyMs) ? latencyMs : null;
 
-function prioritizeRelays(relays, config, nowMs = Date.now()) {
-  const minActive = Math.max(1, Math.min(config.minActiveRelayPool, relays.length || 1));
-  const ranked = rankRelaysByHealth([...new Set(relays)], config, nowMs);
-  const active = ranked.filter((entry) => !entry.summary.quarantined);
-  const quarantined = ranked
-    .filter((entry) => entry.summary.quarantined)
-    .sort((a, b) => a.summary.quarantineRemainingMs - b.summary.quarantineRemainingMs);
-
-  const selected = [...active];
-  if (selected.length < minActive) {
-    const additionalNeeded = Math.min(minActive - selected.length, quarantined.length);
-    selected.push(...quarantined.slice(0, additionalNeeded));
-  }
-
-  return {
-    prioritized: selected.map((entry) => entry.relay),
-    ranked,
-  };
-}
-
-function recordRelayOutcome(relay, success, errorMessage, latencyMs, config, nowMs = Date.now()) {
-  const metrics = ensureRelayMetrics(relay, config);
-  const timedOut = String(errorMessage || '').toLowerCase().includes('timeout')
-    || String(errorMessage || '').toLowerCase().includes('timed out');
-  metrics.recentOutcomes.push({ success, timedOut, latencyMs, atMs: nowMs });
-  if (metrics.recentOutcomes.length > config.rollingWindowSize) {
-    metrics.recentOutcomes.splice(0, metrics.recentOutcomes.length - config.rollingWindowSize);
-  }
-  metrics.lastResultAt = nowMs;
-  metrics.lastLatencyMs = Number.isFinite(latencyMs) ? latencyMs : null;
-
-  if (success) {
-    metrics.failureStreak = 0;
-    if (metrics.quarantineUntil > nowMs) {
-      metrics.quarantineUntil = 0;
-      metrics.cooldownMs = config.quarantineCooldownMs;
+    if (success) {
+      metrics.failureStreak = 0;
+      if (metrics.quarantineUntil > nowMs) {
+        metrics.quarantineUntil = 0;
+        metrics.cooldownMs = config.quarantineCooldownMs;
+      }
+      return;
+    }
+
+    metrics.failureStreak += 1;
+    if (metrics.failureStreak >= config.failureThreshold) {
+      metrics.quarantineUntil = nowMs + metrics.cooldownMs;
+      metrics.cooldownMs = Math.min(config.maxQuarantineCooldownMs, Math.floor(metrics.cooldownMs * 1.5));
     }
-    return;
   }
 
-  metrics.failureStreak += 1;
-  if (metrics.failureStreak >= config.failureThreshold) {
-    metrics.quarantineUntil = nowMs + metrics.cooldownMs;
-    metrics.cooldownMs = Math.min(config.maxQuarantineCooldownMs, Math.floor(metrics.cooldownMs * 1.5));
+  collectSnapshot(relays, config, nowMs = Date.now()) {
+    const uniqueRelays = [...new Set(relays)];
+    return this.rankRelays(uniqueRelays, config, nowMs).map((entry) => ({
+      relay: entry.relay,
+      score: Number(entry.score.toFixed(4)),
+      ...entry.summary,
+    }));
+  }
+
+  maybeLogSnapshot(relays, config, logger, reason, force = false, nowMs = Date.now()) {
+    const intervalReached = nowMs - this.lastSnapshotAt >= config.snapshotIntervalMs;
+    if (!force && !intervalReached) return;
+    this.lastSnapshotAt = nowMs;
+    logger(JSON.stringify({
+      event: 'relay_health_snapshot',
+      reason,
+      relays: this.collectSnapshot(relays, config, nowMs),
+    }));
+  }
+
+  reset() {
+    this.metricsByRelay.clear();
+    this.lastSnapshotAt = 0;
   }
 }
 
-function collectHealthSnapshot(relays, config, nowMs = Date.now()) {
-  const uniqueRelays = [...new Set(relays)];
-  return rankRelaysByHealth(uniqueRelays, config, nowMs).map((entry) => ({
-    relay: entry.relay,
-    score: Number(entry.score.toFixed(4)),
-    ...entry.summary,
-  }));
-}
-
-function maybeLogHealthSnapshot(relays, config, logger, reason, force = false, nowMs = Date.now()) {
-  const intervalReached = nowMs - relayHealthState.lastSnapshotAt >= config.snapshotIntervalMs;
-  if (!force && !intervalReached) return;
-  relayHealthState.lastSnapshotAt = nowMs;
-  logger(JSON.stringify({
-    event: 'relay_health_snapshot',
-    reason,
-    relays: collectHealthSnapshot(relays, config, nowMs),
-  }));
-}
+export const defaultHealthManager = new RelayHealthManager();
 
 function buildRelayHealthConfig(deps) {
   return {
@@ -302,19 +311,20 @@ export async function queryLocks(relays, cadence, dateStr, namespace, deps = {})
     getMinActiveRelayPoolFn = getMinActiveRelayPool,
     errorLogger = console.error,
     healthLogger = console.error,
+    healthManager = defaultHealthManager,
   } = deps;
 
   const pool = poolFactory();
   const tagFilter = `${namespace}-lock-${cadence}-${dateStr}`;
-  const queryTimeoutMs = getQueryTimeoutMsFn();
-  const fallbackRelays = getRelayFallbacksFn().filter((relay) => !relays.includes(relay));
+  const queryTimeoutMs = await getQueryTimeoutMsFn();
+  const fallbackRelays = (await getRelayFallbacksFn()).filter((relay) => !relays.includes(relay));
   const healthConfig = buildRelayHealthConfig({
     ...deps,
-    minActiveRelayPool: getMinActiveRelayPoolFn(),
+    minActiveRelayPool: await getMinActiveRelayPoolFn(),
   });
 
   const runQuery = async (relaySet, phase) => {
-    const { prioritized } = prioritizeRelays(relaySet, healthConfig);
+    const { prioritized } = healthManager.prioritizeRelays(relaySet, healthConfig);
     if (prioritized.length > 0) {
       errorLogger(`[${phase}] Querying ${prioritized.length} relays (${relayListLabel(prioritized)})...`);
     }
@@ -331,14 +341,14 @@ export async function queryLocks(relays, cadence, dateStr, namespace, deps = {})
       );
       const elapsedMs = Date.now() - startedAtMs;
       for (const relay of prioritized) {
-        recordRelayOutcome(relay, true, null, elapsedMs, healthConfig);
+        healthManager.recordOutcome(relay, true, null, elapsedMs, healthConfig);
       }
       return filterActiveLocks(events.map(parseLockEvent));
     } catch (err) {
       const elapsedMs = Date.now() - startedAtMs;
       const message = err instanceof Error ? err.message : String(err);
       for (const relay of prioritized) {
-        recordRelayOutcome(relay, false, message, elapsedMs, healthConfig);
+        healthManager.recordOutcome(relay, false, message, elapsedMs, healthConfig);
       }
       throw new Error(
         `[${phase}] Relay query failed (timeout=${queryTimeoutMs}ms, relays=${relayListLabel(prioritized)}): ${message}`,
@@ -350,12 +360,12 @@ export async function queryLocks(relays, cadence, dateStr, namespace, deps = {})
   const allRelays = mergeRelayList(relays, fallbackRelays);
 
   try {
-    maybeLogHealthSnapshot(allRelays, healthConfig, healthLogger, 'query:periodic');
+    healthManager.maybeLogSnapshot(allRelays, healthConfig, healthLogger, 'query:periodic');
     try {
       return await runQuery(relays, 'query:primary');
     } catch (primaryErr) {
       if (!fallbackRelays.length) {
-        maybeLogHealthSnapshot(allRelays, healthConfig, healthLogger, 'query:failure', true);
+        healthManager.maybeLogSnapshot(allRelays, healthConfig, healthLogger, 'query:failure', true);
         throw primaryErr;
       }
       errorLogger(`WARN: ${primaryErr.message}`);
@@ -398,65 +408,104 @@ async function publishToRelays(pool, relays, event, publishTimeoutMs, phase) {
   });
 }
 
-export async function publishLock(relays, event, deps = {}) {
-  const {
-    poolFactory = () => new SimplePool(),
-    getPublishTimeoutMsFn = getPublishTimeoutMs,
-    getMinSuccessfulRelayPublishesFn = getMinSuccessfulRelayPublishes,
-    getRelayFallbacksFn = getRelayFallbacks,
-    getMinActiveRelayPoolFn = getMinActiveRelayPool,
-    retryAttempts = 4,
-    retryBaseDelayMs = 500,
-    retryCapDelayMs = 8_000,
-    sleepFn = (ms) => new Promise((resolve) => setTimeout(resolve, ms)),
-    randomFn = Math.random,
-    telemetryLogger = console.error,
-    healthLogger = console.error,
-    diagnostics = {},
-  } = deps;
+class LockPublisher {
+  constructor(relays, event, deps) {
+    this.relays = relays;
+    this.event = event;
+    this.deps = deps;
 
-  const correlationId = diagnostics.correlationId || process.env.SCHEDULER_LOCK_CORRELATION_ID || 'none';
-  const attemptId = diagnostics.attemptId || process.env.SCHEDULER_LOCK_ATTEMPT_ID || '1';
+    const {
+      poolFactory = () => new SimplePool(),
+      getPublishTimeoutMsFn = getPublishTimeoutMs,
+      getMinSuccessfulRelayPublishesFn = getMinSuccessfulRelayPublishes,
+      getRelayFallbacksFn = getRelayFallbacks,
+      getMinActiveRelayPoolFn = getMinActiveRelayPool,
+      retryAttempts = 4,
+      retryBaseDelayMs = 500,
+      retryCapDelayMs = 8_000,
+      sleepFn = (ms) => new Promise((resolve) => setTimeout(resolve, ms)),
+      randomFn = Math.random,
+      telemetryLogger = console.error,
+      healthLogger = console.error,
+      diagnostics = {},
+    } = deps;
 
-  const pool = poolFactory();
-  const publishTimeoutMs = getPublishTimeoutMsFn();
-  const minSuccesses = getMinSuccessfulRelayPublishesFn();
-  const fallbackRelays = getRelayFallbacksFn().filter((relay) => !relays.includes(relay));
-  const maxAttempts = Math.max(1, Math.floor(retryAttempts));
-  const healthConfig = buildRelayHealthConfig({
-    ...deps,
-    minActiveRelayPool: getMinActiveRelayPoolFn(),
-  });
+    this.pool = poolFactory();
+    this.publishTimeoutMs = getPublishTimeoutMsFn();
+    this.minSuccesses = getMinSuccessfulRelayPublishesFn();
+    this.fallbackRelays = getRelayFallbacksFn().filter((relay) => !relays.includes(relay));
+    this.maxAttempts = Math.max(1, Math.floor(retryAttempts));
+    this.healthConfig = buildRelayHealthConfig({
+      ...deps,
+      minActiveRelayPool: getMinActiveRelayPoolFn(),
+    });
 
-  const publishOnce = async () => {
+    this.retryBaseDelayMs = retryBaseDelayMs;
+    this.retryCapDelayMs = retryCapDelayMs;
+    this.sleepFn = sleepFn;
+    this.randomFn = randomFn;
+    this.telemetryLogger = telemetryLogger;
+    this.healthLogger = healthLogger;
+
+    this.correlationId = diagnostics.correlationId || process.env.SCHEDULER_LOCK_CORRELATION_ID || 'none';
+    this.attemptId = diagnostics.attemptId || process.env.SCHEDULER_LOCK_ATTEMPT_ID || '1';
+
+    this.allRelays = mergeRelayList(relays, this.fallbackRelays);
+  }
+
+  async publish() {
+    try {
+      maybeLogHealthSnapshot(this.allRelays, this.healthConfig, this.healthLogger, 'publish:periodic');
+      let lastAttemptState = null;
+      const retryTimeline = [];
+      const overallStartedAt = Date.now();
+      let terminalFailureCategory = 'relay_publish_non_retryable';
+
+      for (let attemptNumber = 1; attemptNumber <= this.maxAttempts; attemptNumber += 1) {
+        const startedAtMs = Date.now();
+        lastAttemptState = await this.executePublishCycle();
+        const elapsedMs = Date.now() - startedAtMs;
+        retryTimeline.push({
+          publishAttempt: attemptNumber,
+          successCount: lastAttemptState.successCount,
+          relayAttemptedCount: lastAttemptState.attempted.size,
+          elapsedMs,
+        });
+
+        if (lastAttemptState.successCount >= this.minSuccesses) {
+          this.logSuccess(attemptNumber, lastAttemptState, retryTimeline, overallStartedAt);
+          return this.event;
+        }
+
+        const { hasTransientFailure, hasPermanentFailure } = this.analyzeFailures(lastAttemptState.failures);
+        const canRetry = attemptNumber < this.maxAttempts && hasTransientFailure && !hasPermanentFailure;
+
+        terminalFailureCategory = this.determineFailureCategory(hasTransientFailure, hasPermanentFailure, attemptNumber);
+
+        if (!canRetry) {
+          break;
+        }
+
+        const nextDelayMs = calculateBackoffDelayMs(attemptNumber, this.retryBaseDelayMs, this.retryCapDelayMs, this.randomFn);
+        this.logRetry(attemptNumber, lastAttemptState.failures, elapsedMs, nextDelayMs);
+        await this.sleepFn(nextDelayMs);
+      }
+
+      this.handleFinalFailure(lastAttemptState, terminalFailureCategory, retryTimeline, overallStartedAt);
+    } finally {
+      this.pool.close(this.allRelays);
+    }
+  }
+
+  async executePublishCycle() {
     const attempted = new Set();
     const publishResults = [];
 
-    const attemptPhase = async (phaseRelays, phaseName) => {
-      if (!phaseRelays.length) return;
-      const { prioritized } = prioritizeRelays(phaseRelays, healthConfig);
-      if (!prioritized.length) return;
-
-      console.error(`[${phaseName}] Publishing to ${prioritized.length} relays (${prioritized.join(', ')})...`);
-
-      const startedAtMs = Date.now();
-      for (const relay of prioritized) {
-        attempted.add(relay);
-      }
-      const phaseResults = await publishToRelays(pool, prioritized, event, publishTimeoutMs, phaseName);
-      const elapsedMs = Date.now() - startedAtMs;
-      for (const result of phaseResults) {
-        result.latencyMs = elapsedMs;
-        recordRelayOutcome(result.relay, result.success, result.message, elapsedMs, healthConfig);
-      }
-      publishResults.push(...phaseResults);
-    };
-
-    await attemptPhase(relays, 'publish:primary');
+    await this.attemptPhase(this.relays, 'publish:primary', attempted, publishResults);
 
     let successCount = publishResults.filter((result) => result.success).length;
-    if (successCount < minSuccesses && fallbackRelays.length > 0) {
-      await attemptPhase(fallbackRelays, 'publish:fallback');
+    if (successCount < this.minSuccesses && this.fallbackRelays.length > 0) {
+      await this.attemptPhase(this.fallbackRelays, 'publish:fallback', attempted, publishResults);
       successCount = publishResults.filter((result) => result.success).length;
     }
 
@@ -468,89 +517,88 @@ export async function publishLock(relays, event, deps = {}) {
       }));
 
     return { attempted, publishResults, successCount, failures };
-  };
+  }
 
-  const allRelays = mergeRelayList(relays, fallbackRelays);
+  async attemptPhase(phaseRelays, phaseName, attempted, publishResults) {
+    if (!phaseRelays.length) return;
+    const { prioritized } = prioritizeRelays(phaseRelays, this.healthConfig);
+    if (!prioritized.length) return;
 
-  try {
-    maybeLogHealthSnapshot(allRelays, healthConfig, healthLogger, 'publish:periodic');
-    let lastAttemptState = null;
-    const retryTimeline = [];
-    const overallStartedAt = Date.now();
-    let terminalFailureCategory = 'relay_publish_non_retryable';
-    for (let attemptNumber = 1; attemptNumber <= maxAttempts; attemptNumber += 1) {
-      const startedAtMs = Date.now();
-      lastAttemptState = await publishOnce();
-      const elapsedMs = Date.now() - startedAtMs;
-      retryTimeline.push({
-        publishAttempt: attemptNumber,
-        successCount: lastAttemptState.successCount,
-        relayAttemptedCount: lastAttemptState.attempted.size,
-        elapsedMs,
-      });
+    console.error(`[${phaseName}] Publishing to ${prioritized.length} relays (${prioritized.join(', ')})...`);
 
-      if (lastAttemptState.successCount >= minSuccesses) {
-        telemetryLogger(JSON.stringify({
-          event: 'lock_publish_quorum_met',
-          correlationId,
-          attemptId,
-          publishAttempt: attemptNumber,
-          successCount: lastAttemptState.successCount,
-          relayAttemptedCount: lastAttemptState.attempted.size,
-          requiredSuccesses: minSuccesses,
-          timeoutMs: publishTimeoutMs,
-          retryTimeline,
-          totalElapsedMs: Date.now() - overallStartedAt,
-        }));
-        console.error(
-          `  Published to ${lastAttemptState.successCount}/${lastAttemptState.attempted.size} relays `
-          + `(required=${minSuccesses}, timeout=${publishTimeoutMs}ms)`,
-        );
-        return event;
-      }
-
-      const hasTransientFailure = lastAttemptState.failures.some((failure) => isTransientPublishCategory(failure.category));
-      const hasPermanentFailure = lastAttemptState.failures.some((failure) => !isTransientPublishCategory(failure.category));
-      const canRetry = attemptNumber < maxAttempts && hasTransientFailure && !hasPermanentFailure;
-
-      if (hasTransientFailure && !hasPermanentFailure && attemptNumber >= maxAttempts) {
-        terminalFailureCategory = 'relay_publish_quorum_failure';
-      } else if (hasPermanentFailure) {
-        terminalFailureCategory = 'relay_publish_non_retryable';
-      } else {
-        terminalFailureCategory = 'relay_publish_quorum_failure';
-      }
-
-      if (!canRetry) {
-        break;
-      }
-
-      const nextDelayMs = calculateBackoffDelayMs(attemptNumber, retryBaseDelayMs, retryCapDelayMs, randomFn);
-      for (const failure of lastAttemptState.failures) {
-        if (!isTransientPublishCategory(failure.category)) continue;
-        telemetryLogger(JSON.stringify({
-          event: 'lock_publish_retry',
-          correlationId,
-          attemptId,
-          publishAttempt: attemptNumber,
-          relayUrl: failure.relay,
-          errorCategory: failure.category,
-          elapsedMs,
-          nextDelayMs,
-        }));
-      }
-
-      await sleepFn(nextDelayMs);
+    const startedAtMs = Date.now();
+    for (const relay of prioritized) {
+      attempted.add(relay);
     }
+    const phaseResults = await publishToRelays(this.pool, prioritized, this.event, this.publishTimeoutMs, phaseName);
+    const elapsedMs = Date.now() - startedAtMs;
+    for (const result of phaseResults) {
+      result.latencyMs = elapsedMs;
+      recordRelayOutcome(result.relay, result.success, result.message, elapsedMs, this.healthConfig);
+    }
+    publishResults.push(...phaseResults);
+  }
 
+  analyzeFailures(failures) {
+    const hasTransientFailure = failures.some((failure) => isTransientPublishCategory(failure.category));
+    const hasPermanentFailure = failures.some((failure) => !isTransientPublishCategory(failure.category));
+    return { hasTransientFailure, hasPermanentFailure };
+  }
+
+  determineFailureCategory(hasTransientFailure, hasPermanentFailure, attemptNumber) {
+    if (hasTransientFailure && !hasPermanentFailure && attemptNumber >= this.maxAttempts) {
+      return 'relay_publish_quorum_failure';
+    } else if (hasPermanentFailure) {
+      return 'relay_publish_non_retryable';
+    } else {
+      return 'relay_publish_quorum_failure';
+    }
+  }
+
+  logSuccess(attemptNumber, lastAttemptState, retryTimeline, overallStartedAt) {
+    this.telemetryLogger(JSON.stringify({
+      event: 'lock_publish_quorum_met',
+      correlationId: this.correlationId,
+      attemptId: this.attemptId,
+      publishAttempt: attemptNumber,
+      successCount: lastAttemptState.successCount,
+      relayAttemptedCount: lastAttemptState.attempted.size,
+      requiredSuccesses: this.minSuccesses,
+      timeoutMs: this.publishTimeoutMs,
+      retryTimeline,
+      totalElapsedMs: Date.now() - overallStartedAt,
+    }));
+    console.error(
+      `  Published to ${lastAttemptState.successCount}/${lastAttemptState.attempted.size} relays `
+      + `(required=${this.minSuccesses}, timeout=${this.publishTimeoutMs}ms)`,
+    );
+  }
+
+  logRetry(attemptNumber, failures, elapsedMs, nextDelayMs) {
+    for (const failure of failures) {
+      if (!isTransientPublishCategory(failure.category)) continue;
+      this.telemetryLogger(JSON.stringify({
+        event: 'lock_publish_retry',
+        correlationId: this.correlationId,
+        attemptId: this.attemptId,
+        publishAttempt: attemptNumber,
+        relayUrl: failure.relay,
+        errorCategory: failure.category,
+        elapsedMs,
+        nextDelayMs,
+      }));
+    }
+  }
+
+  handleFinalFailure(lastAttemptState, terminalFailureCategory, retryTimeline, overallStartedAt) {
     const reasonDistribution = {};
     const failureLines = lastAttemptState.failures
       .map((result) => {
         reasonDistribution[result.category] = (reasonDistribution[result.category] || 0) + 1;
-        telemetryLogger(JSON.stringify({
+        this.telemetryLogger(JSON.stringify({
           event: 'lock_publish_failure',
-          correlationId,
-          attemptId,
+          correlationId: this.correlationId,
+          attemptId: this.attemptId,
           relayUrl: result.relay,
           phase: result.phase,
           reason: result.category,
@@ -559,34 +607,35 @@ export async function publishLock(relays, event, deps = {}) {
         return `${result.relay} (${result.phase}, reason=${result.category}): ${result.message}`;
       });
 
-    telemetryLogger(JSON.stringify({
+    this.telemetryLogger(JSON.stringify({
       event: 'lock_publish_quorum_failed',
-      correlationId,
-      attemptId,
+      correlationId: this.correlationId,
+      attemptId: this.attemptId,
       errorCategory: terminalFailureCategory,
       successCount: lastAttemptState.successCount,
       relayAttemptedCount: lastAttemptState.attempted.size,
-      requiredSuccesses: minSuccesses,
-      timeoutMs: publishTimeoutMs,
-      attempts: maxAttempts,
+      requiredSuccesses: this.minSuccesses,
+      timeoutMs: this.publishTimeoutMs,
+      attempts: this.maxAttempts,
       reasonDistribution,
       retryTimeline,
       totalElapsedMs: Date.now() - overallStartedAt,
     }));
 
-    maybeLogHealthSnapshot(allRelays, healthConfig, healthLogger, 'publish:failure', true);
+    maybeLogHealthSnapshot(this.allRelays, this.healthConfig, this.healthLogger, 'publish:failure', true);
     throw new Error(
       `Failed relay publish quorum in publish phase: ${lastAttemptState.successCount}/${lastAttemptState.attempted.size} successful `
-      + `(required=${minSuccesses}, timeout=${publishTimeoutMs}ms, attempts=${maxAttempts}, attempt_id=${attemptId}, correlation_id=${correlationId}, error_category=${terminalFailureCategory}, total_retry_timeline_ms=${Date.now() - overallStartedAt})\n`
+      + `(required=${this.minSuccesses}, timeout=${this.publishTimeoutMs}ms, attempts=${this.maxAttempts}, attempt_id=${this.attemptId}, correlation_id=${this.correlationId}, error_category=${terminalFailureCategory}, total_retry_timeline_ms=${Date.now() - overallStartedAt})\n`
       + `  retry timeline: ${retryTimeline.map((item) => `#${item.publishAttempt}:${item.elapsedMs}ms`).join(', ')}\n`
       + `  ${failureLines.join('\n  ')}`,
     );
-  } finally {
-    pool.close(allRelays);
   }
 }
 
+export async function publishLock(relays, event, deps = {}) {
+  return new LockPublisher(relays, event, deps).publish();
+}
+
 export function _resetRelayHealthState() {
-  relayHealthState.metricsByRelay.clear();
-  relayHealthState.lastSnapshotAt = 0;
+  defaultHealthManager.reset();
 }

src/perf/benchmark-config-load.mjs

@@ -0,0 +1,45 @@
+import { loadTorchConfig, _resetTorchConfigCache } from '../torch-config.mjs';
+import fs from 'node:fs';
+import path from 'node:path';
+
+const configPath = path.resolve(process.cwd(), 'torch-config.json');
+const originalConfig = fs.existsSync(configPath) ? fs.readFileSync(configPath) : null;
+
+// Ensure a config file exists for benchmark
+if (!fs.existsSync(configPath)) {
+  fs.writeFileSync(configPath, JSON.stringify({
+    nostrLock: {
+      relays: ["wss://relay.damus.io"],
+      namespace: "test-ns"
+    }
+  }, null, 2));
+}
+
+const ITERATIONS = 1000;
+const start = process.hrtime.bigint();
+
+async function run() {
+    for (let i = 0; i < ITERATIONS; i++) {
+      _resetTorchConfigCache();
+      try {
+        const res = await loadTorchConfig();
+        // Use res to prevent dead code elimination if JIT is smart (though side effects exist)
+        if (!res) throw new Error("Failed");
+      } catch (e) {
+        // ignore
+      }
+    }
+
+    const end = process.hrtime.bigint();
+    const duration = Number(end - start) / 1e6; // ms
+
+    console.log(`[ASYNC] Loaded config ${ITERATIONS} times in ${duration.toFixed(2)}ms`);
+    console.log(`[ASYNC] Average: ${(duration / ITERATIONS).toFixed(4)}ms per load`);
+
+    // Cleanup
+    if (!originalConfig && fs.existsSync(configPath)) {
+      fs.unlinkSync(configPath);
+    }
+}
+
+run();

src/relay-health.mjs

@@ -196,15 +196,22 @@ async function appendHistory(historyPath, entry) {
 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;
 
+  let {
+    relays,
+    namespace,
+    historyPath,
+  } = options;
+
+  if (!relays) relays = await getRelays();
+  if (!namespace) namespace = await getNamespace();
+  if (!historyPath) historyPath = path.resolve(process.cwd(), 'task-logs', 'relay-health', `${cadence}.jsonl`);
+
   const relayResults = [];
   for (const relay of relays) {
     const relayResult = {

src/roster.mjs

@@ -94,8 +94,8 @@ function parseEnvRoster(value) {
     .filter(Boolean);
 }
 
-export function getRoster(cadence) {
-  const config = loadTorchConfig();
+export async function getRoster(cadence) {
+  const config = await loadTorchConfig();
   const dailyFromEnv = parseEnvRoster(process.env.NOSTR_LOCK_DAILY_ROSTER);
   const weeklyFromEnv = parseEnvRoster(process.env.NOSTR_LOCK_WEEKLY_ROSTER);
   const canonical = loadCanonicalRoster();

src/services/memory/index.js

@@ -198,10 +198,15 @@ export async function ingestEvents(events, options = {}) {
  * @returns {Promise<import('./schema.js').MemoryRecord[]>} Sorted list of relevant memories.
  */
 export async function getRelevantMemories(params) {
-  const { repository = memoryRepository, ...queryParams } = params;
+  const {
+    repository = memoryRepository,
+    ranker = filterAndRankMemories,
+    cache: cacheProvider = cache,
+    ...queryParams
+  } = params;
   const telemetry = buildTelemetryEmitter(params);
   const cacheKey = JSON.stringify(queryParams);
-  const cached = cache.get(cacheKey);
+  const cached = cacheProvider.get(cacheKey);
   if (cached) {
     telemetry('memory:retrieved', {
       agent_id: queryParams.agent_id,
@@ -220,9 +225,9 @@ export async function getRelevantMemories(params) {
     ? await repository.listMemories(queryParams)
     : [...memoryStore.values()];
 
-  const ranked = await filterAndRankMemories(source, queryParams);
+  const ranked = await ranker(source, queryParams);
   await updateMemoryUsage(repository, ranked.map((memory) => memory.id));
-  cache.set(cacheKey, ranked);
+  cacheProvider.set(cacheKey, ranked);
 
   telemetry('memory:retrieved', {
     agent_id: queryParams.agent_id,

src/torch-config.mjs

@@ -127,18 +127,20 @@ export function _resetTorchConfigCache() {
   cachedConfig = null;
 }
 
-export function loadTorchConfig(fileSystem = fs) {
+export async function loadTorchConfig(fileSystem = fs) {
   if (cachedConfig) return cachedConfig;
 
   const configPath = getTorchConfigPath();
   let raw = {};
 
-  if (fileSystem.existsSync(configPath)) {
-    try {
-      raw = JSON.parse(fileSystem.readFileSync(configPath, 'utf8'));
-    } catch (err) {
+  try {
+    const content = await fileSystem.promises.readFile(configPath, 'utf8');
+    raw = JSON.parse(content);
+  } catch (err) {
+    if (err.code !== 'ENOENT') {
       throw new Error(`Failed to parse ${configPath}: ${err.message}`, { cause: err });
     }
+    // If file missing (ENOENT), we use empty object defaults
   }
 
   cachedConfig = parseTorchConfig(raw, configPath);
@@ -199,8 +201,8 @@ function resolveIntegerConfig(envKey, configValue, defaultValue, validator, conf
   return value;
 }
 
-export function getRelays() {
-  const config = loadTorchConfig();
+export async function getRelays() {
+  const config = await loadTorchConfig();
   const envRelays = process.env.NOSTR_LOCK_RELAYS;
   if (envRelays) {
     return parseEnvRelayList(envRelays, 'NOSTR_LOCK_RELAYS');
@@ -214,8 +216,8 @@ export function getRelays() {
   return DEFAULT_RELAYS;
 }
 
-export function getRelayFallbacks() {
-  const config = loadTorchConfig();
+export async function getRelayFallbacks() {
+  const config = await loadTorchConfig();
   const envRelays = process.env.NOSTR_LOCK_RELAY_FALLBACKS;
   if (envRelays) {
     return parseEnvRelayList(envRelays, 'NOSTR_LOCK_RELAY_FALLBACKS');
@@ -223,14 +225,14 @@ export function getRelayFallbacks() {
   return config.nostrLock.relayFallbacks || [];
 }
 
-export function getNamespace() {
-  const config = loadTorchConfig();
+export async function getNamespace() {
+  const config = await loadTorchConfig();
   const namespace = (process.env.NOSTR_LOCK_NAMESPACE || config.nostrLock.namespace || DEFAULT_NAMESPACE).trim();
   return namespace || DEFAULT_NAMESPACE;
 }
 
-export function getTtl() {
-  const config = loadTorchConfig();
+export async function getTtl() {
+  const config = await loadTorchConfig();
   const envTtl = process.env.NOSTR_LOCK_TTL;
   if (envTtl) {
     const parsed = parseInt(envTtl, 10);
@@ -242,8 +244,8 @@ export function getTtl() {
   return DEFAULT_TTL;
 }
 
-export function getQueryTimeoutMs() {
-  const config = loadTorchConfig();
+export async function getQueryTimeoutMs() {
+  const config = await loadTorchConfig();
   return resolveIntegerConfig(
     'NOSTR_LOCK_QUERY_TIMEOUT_MS',
     config.nostrLock.queryTimeoutMs,
@@ -253,8 +255,8 @@ export function getQueryTimeoutMs() {
   );
 }
 
-export function getPublishTimeoutMs() {
-  const config = loadTorchConfig();
+export async function getPublishTimeoutMs() {
+  const config = await loadTorchConfig();
   return resolveIntegerConfig(
     'NOSTR_LOCK_PUBLISH_TIMEOUT_MS',
     config.nostrLock.publishTimeoutMs,
@@ -264,8 +266,8 @@ export function getPublishTimeoutMs() {
   );
 }
 
-export function getMinSuccessfulRelayPublishes() {
-  const config = loadTorchConfig();
+export async function getMinSuccessfulRelayPublishes() {
+  const config = await loadTorchConfig();
   return resolveIntegerConfig(
     'NOSTR_LOCK_MIN_SUCCESSFUL_PUBLISHES',
     config.nostrLock.minSuccessfulRelayPublishes,
@@ -275,8 +277,8 @@ export function getMinSuccessfulRelayPublishes() {
   );
 }
 
-export function getMinActiveRelayPool() {
-  const config = loadTorchConfig();
+export async function getMinActiveRelayPool() {
+  const config = await loadTorchConfig();
   return resolveIntegerConfig(
     'NOSTR_LOCK_MIN_ACTIVE_RELAY_POOL',
     config.nostrLock.minActiveRelayPool,
@@ -286,8 +288,8 @@ export function getMinActiveRelayPool() {
   );
 }
 
-export function getHashtag() {
-  const config = loadTorchConfig();
+export async function getHashtag() {
+  const config = await loadTorchConfig();
   const envValue = process.env.NOSTR_LOCK_HASHTAG;
   if (envValue) {
     return envValue.trim();
@@ -295,12 +297,12 @@ export function getHashtag() {
   if (config.dashboard.hashtag) {
     return config.dashboard.hashtag;
   }
-  const namespace = getNamespace();
+  const namespace = await getNamespace();
   return `${namespace}-agent-lock`;
 }
 
-export function getDashboardAuth() {
-  const config = loadTorchConfig();
+export async function getDashboardAuth() {
+  const config = await loadTorchConfig();
   const envValue = process.env.TORCH_DASHBOARD_AUTH;
   if (envValue) {
     return envValue.trim();