fix(pipeline): baseline double-count, counter drift, retest totals, loop stop

From the bug hunt, adversarially reviewed (safe to commit, no CRITICAL/HIGH):

- baselineHistory double-count: refreshBaseline() ran per node ATTEMPT (before
  testNode), so a stop+resume or pause-retry of one node pushed multiple baseline
  samples (user saw baseline=4 vs speed=3 for 3 nodes). Now refreshBaseline only
  stashes a pending sample; commitBaselineSample(addr) pushes once per RECORDED
  node (dedup by address, per-invocation closure), at the success/fail result
  sites — interrupted/retried attempts commit nothing. baseline == tested-count.
- runAudit now calls recomputeCounters(state) at completion (only on finish/stop,
  not the error path) so persisted counters match results[] instead of the
  drift-prone live increments.
- Iron-Rule retest block now guards the success flip on result.actualMbps!=null
  (mirrors the internet-recovery block); a null-mbps result persists as a fail
  without a bogus failed--/tested++.
- runRetestSkips no longer rewrites totalNodes = tested+failed (which collapsed
  the grand total for partial/stopped runs) — uses Math.max(results.length,total).
- fresh runAudit now resets skippedNodes alongside the sibling counters.
- continuous pause()/stop() now propagate to the in-flight pipeline
  (triggerPipelineStop() + loopState.stopRequested) so a sweep and its payments
  halt promptly instead of running to completion first.

Test suite green (188/31/45/35).

Author: Sentinel-Bluebuilder

audit/continuous.js

@@ -23,7 +23,7 @@ import { EventEmitter } from 'events';
 import { writeFileSync, readFileSync, existsSync, mkdirSync } from 'fs';
 import path from 'path';
 import { fileURLToPath } from 'url';
-import { createState, setActiveDbRunId, getActiveDbRunId } from './pipeline.js';
+import { createState, setActiveDbRunId, getActiveDbRunId, triggerPipelineStop } from './pipeline.js';
 import { sleep } from '../protocol/speedtest.js';
 
 const __dirname = path.dirname(fileURLToPath(import.meta.url));
@@ -144,6 +144,10 @@ const _ctrl = {
   paused: false,
   pausedBatch: null, // { batchId, mode, planId, subscriptionId, subscriptionGranter, frozenNodes, testedAddrs }
   resumeIntent: null, // { batchId, frozenNodes, testedAddrs } — set by resume(), consumed by _runLoop
+  // The pipeline state object for the currently in-flight pass. Held so pause()/stop()
+  // can raise its stopRequested flag and abort the running runAudit/runSubPlanTest sweep
+  // promptly, instead of letting the whole pass (and its payments) finish first.
+  loopState: null,
 };
 
 // Injected pipeline runner (overridable in tests)
@@ -371,6 +375,8 @@ async function _runLoop() {
       // without this, every continuous-loop result fell back to sdk:'js'.
       const loopState = createState();
       loopState.stopRequested = false;
+      // Expose the in-flight state so pause()/stop() can abort the running pipeline.
+      _ctrl.loopState = loopState;
       if (_ctrl.activeSDK)   loopState.activeSDK   = _ctrl.activeSDK;
       if (_ctrl.pricingMode) loopState.pricingMode = _ctrl.pricingMode;
       loopState.runMode             = _ctrl.mode || 'p2p';
@@ -613,6 +619,7 @@ async function _runLoop() {
   } finally {
     _ctrl.running = false;
     _ctrl.stopRequested = false;
+    _ctrl.loopState = null; // no in-flight pass anymore
     _persistLoopConfig();
     _emitScoped('loop:stopped', {
       iterations: _ctrl.iteration,
@@ -728,6 +735,10 @@ export async function start(opts = {}) {
 export function stop() {
   if (!_ctrl.running) return { ok: true, alreadyStopped: true };
   _ctrl.stopRequested = true;
+  // Propagate to the in-flight pipeline so the running runAudit/runSubPlanTest sweep
+  // (and its payments) returns promptly instead of completing the whole pass first.
+  if (_ctrl.loopState) _ctrl.loopState.stopRequested = true;
+  triggerPipelineStop();
   _emitScoped('loop:stopping', {});
   return { ok: true };
 }
@@ -744,6 +755,11 @@ export function pause() {
   if (_ctrl.paused) return { ok: false, error: 'already paused' };
   // Raise stopRequested so the in-flight pipeline returns.
   _ctrl.stopRequested = true;
+  // Propagate to the in-flight pipeline state + wake its stop-aware sleeps so the
+  // running sweep aborts promptly, rather than finishing the full pass (and its
+  // payments) before the pause takes effect — matching this function's docstring.
+  if (_ctrl.loopState) _ctrl.loopState.stopRequested = true;
+  triggerPipelineStop();
   // Sentinel for _runLoop: break out without clearing state.
   _ctrl.paused = true;
   _emitScoped('loop:stopping', {});

audit/pipeline.js

@@ -617,6 +617,7 @@ export async function runAudit(resume, state, broadcast, preloadedNodes = null,
     results.length = 0;
     state.testedNodes = 0;
     state.failedNodes = 0;
+    state.skippedNodes = 0;
     state.passed15 = 0;
     state.passed10 = 0;
     state.passedBaseline = 0;
@@ -666,15 +667,32 @@ export async function runAudit(resume, state, broadcast, preloadedNodes = null,
   // ── On-chain reporter (opt-in) ─────────────────────────────────────────
   _initOnchainReporter(account, client, state, broadcast);
 
+  // Latest baseline sample measured for the node currently in-flight. refreshBaseline()
+  // (called once per ATTEMPT) only refreshes the live state.baselineMbps and stashes the
+  // sample here; commitBaselineSample() pushes it into baselineHistory at most once per
+  // node that actually records a result. This keeps the HISTORY append aligned with
+  // processed nodes (1 per result) instead of per attempt — an interrupted/pause-retried
+  // node no longer double-appends and drifts baselineHistory above nodeSpeedHistory.
+  let _pendingBaselineSample = null;
+  let _lastBaselineHistoryAddr = null;
   async function refreshBaseline() {
     try {
       const bl = await speedtestDirect();
       state.baselineMbps = bl.mbps;
       const ts = new Date().toLocaleTimeString('en-US', { hour12: false });
-      state.baselineHistory = [...state.baselineHistory, { mbps: bl.mbps, ts }].slice(-10);
+      _pendingBaselineSample = { mbps: bl.mbps, ts };
       broadcast('state', { state });
     } catch { }
   }
+  // Append the pending baseline sample for `addr` at most once. Deduping by address means
+  // a node re-tested after Stop/Resume or a pause-retry (j--) appends exactly one sample.
+  function commitBaselineSample(addr) {
+    if (!_pendingBaselineSample) return;
+    if (addr && addr === _lastBaselineHistoryAddr) return;
+    state.baselineHistory = [...state.baselineHistory, _pendingBaselineSample].slice(-10);
+    _lastBaselineHistoryAddr = addr || null;
+    _pendingBaselineSample = null;
+  }
 
   // ── Phase 1: Fetch node list ───────────────────────────────────────────
   let allNodes;
@@ -909,6 +927,7 @@ export async function runAudit(resume, state, broadcast, preloadedNodes = null,
         } else {
           state.failedNodes++;
         }
+        commitBaselineSample(node.address); // 1 baseline sample per recorded node
         upsertResult(result); // success — no snippet needed
         state.resumeHeadAddr = null;
         saveResults();
@@ -977,6 +996,7 @@ export async function runAudit(resume, state, broadcast, preloadedNodes = null,
         // FAIL result — zero-skip: explicit failure
         const failResult = buildFailResult(node, status, state, errMsg, error?.diag || {});
         state.failedNodes++;
+        commitBaselineSample(node.address); // 1 baseline sample per recorded node
         upsertResult(failResult, _logSnippet);
         state.resumeHeadAddr = null;
         saveResults();
@@ -1169,14 +1189,23 @@ export async function runAudit(resume, state, broadcast, preloadedNodes = null,
         break;
       }
 
-      if (result) {
+      if (result && result.actualMbps != null) {
+        // Only a real-speed result flips this node from failed → tested.
         state.failedNodes = Math.max(0, state.failedNodes - 1);
         state.testedNodes++;
         if (result.pass10mbps) state.passed10++;
         upsertResult(result);
         saveResults();
         broadcast('result', { result, state });
         broadcast('log', { msg: `  ✓ Retest PASS: ${result.actualMbps} Mbps` });
+      } else if (result) {
+        // Truthy result but null mbps (e.g. SESSION_UNMAPPED) — still a failure (it
+        // was already counted as failed before this retest). Persist without touching
+        // counters; flipping to "tested" here would wrongly clear a still-failed node.
+        upsertResult(result, _sanitizeSnippet(_ironSnippet));
+        saveResults();
+        broadcast('result', { result, state });
+        broadcast('log', { msg: `  ✗ Retest FAIL: ${result.errorCode || 'no speed'}` });
       } else {
         const errMsg = error?.message || 'Unknown';
         const failResult = buildFailResult(node, status, state, errMsg, error?.diag || {});
@@ -1195,6 +1224,10 @@ export async function runAudit(resume, state, broadcast, preloadedNodes = null,
 
   const finalCache = getCacheStats();
   await _finalizeOnchainReporter();
+  // Reconcile the live-incremented counters against results[] so the final
+  // persisted summary matches the authoritative classifier (the per-node
+  // increments above can drift across Stop/Resume, pause-retries and retests).
+  recomputeCounters(state);
   state.status = 'done';
   state.completedAt = new Date().toISOString();
   state.currentNode = null;
@@ -1218,7 +1251,12 @@ export async function runRetestSkips(skipAddrs, state, broadcast) {
   state.retestPassed = 0;
   state.retestFailed = 0;
   recomputeCounters(state);
-  state.totalNodes = state.testedNodes + state.failedNodes; // show grand total
+  // Preserve the run's true node count. Deriving totalNodes from tested+failed
+  // collapses the grand total for a run that was Stopped partway (skipped /
+  // TEST_RUN_SKIP rows are classified as skippedNodes, not failed, so they'd
+  // vanish from the total and break "remaining"). results.length is the real
+  // count of rows; keep whichever is larger so an already-correct chain total wins.
+  state.totalNodes = Math.max(results.length, state.totalNodes || 0); // show grand total
   clearPoisonedSessions();
   clearPaidNodes();
   invalidateSessionCache(); // Force fresh session lookups — prevents stale mappings