dsh_shenxian/test/relay-failover.test.mjs

/**
 * 覆盖网络 · **序⑦ 中继失败切流** 单测 —— "杀掉任一台中继 ⇒ 客户端自动切到另一台"。
 *
 * ## 这个文件要回答的两个问题
 * 1. **候选集还会不会退化成单点？**（改造前 `pickFromDoc` 取到第一个就 `return`
 *    ⇒ 目录里排第二的那台**永远选不中**，重解析一百次拿回来的还是同一个字符串）
 * 2. **"当前这条不健康"时到底会不会换到下一条**，且**换不动时会不会把本来能用的通路打掉**？
 *
 * ## 四条行为断言（对应交接单 §6 E1–E4 / E7–E9）
 * - **候选全取出来**（顺序 = `relays[]` → `bootstrap[]`）；
 * - **`exclude` 生效**，且**不传 `exclude` 时与改造前逐字一致**（D9 向后兼容）；
 * - **不健康判据可读**（`unhealthySinceMs` 非空 / 健康时归零）；
 * - **切换的四条纪律**：能切就切 / 无候选不切空 / 冷却期内不回跳 / 新通道建不起来就保留原通道。
 *   且 `[relay-switch]` 日志行数 **必然等于** `switches` 计数（D7 判别器可断言）。
 *
 * 运行：`node --test test/relay-failover.test.mjs`（Node ≥ 22；测 `lib/` 产物，先 `npm run build`）。
 *
 * @module test/relay-failover
 */

import assert from 'node:assert/strict'
import { generateKeyPairSync, randomBytes } from 'node:crypto'
import { createServer } from 'node:http'
import { createServer as createTcpServer } from 'node:net'
import { test } from 'node:test'
import {
  DIRECTORY_PATH,
  RelayClient,
  RelayFailoverSupervisor,
  RelayServer,
  buildDirectoryDocument,
  gracefulBurstMsDefault,
  listOverlayRelayCandidates,
  openedChannelFailedTerminally,
  resolveOverlayRelay,
  signDirectory,
  waitUpOnStatus,
} from '../lib/net/relay/index.js'
/** 序㉗：候选链观测（`OBS-21` 的判据锚点 —— 行格式一变，探针就会**静默取不到值**）。 */
import { CAND_OBS_PREFIX, RelayCandidateObservation, candidateObsMs } from '../lib/worker/relay-tunnel.js'

/* ─────────── 搭台工具 ─────────── */

const RELAY_PATH = '/dshs-relay'

function makeKeys() {
  const { publicKey, privateKey } = generateKeyPairSync('ed25519')
  return {
    privatePem: privateKey.export({ type: 'pkcs8', format: 'pem' }).toString(),
    publicPem: publicKey.export({ type: 'spki', format: 'pem' }).toString(),
  }
}

/** 一份自签目录：`relays[]` 按给定顺序（**不过滤私网** —— `buildDirectoryDocument` 只做解析/去重）。 */
function signedDoc(relays, keyPem, bootstrap = []) {
  const doc = buildDirectoryDocument({
    relays,
    bootstrap,
    network: 'ops',
    now: Date.now(),
    refreshAfterSeconds: 300,
  })
  return { doc, sig: signDirectory(doc, keyPem) }
}

/** 起一个真的目录端点（`node:http`），请求 `DIRECTORY_PATH` 返回当前 doc+sig。 */
function serveDirectory(initial) {
  const state = { ...initial }
  const server = createServer((req, res) => {
    if (!req.url || !req.url.startsWith(DIRECTORY_PATH)) {
      res.writeHead(404).end('{}')
      return
    }
    res.writeHead(200, { 'content-type': 'application/json', 'cache-control': 'no-store' })
    res.end(JSON.stringify({ ...state.doc, sig: state.sig }))
  })
  return {
    state,
    async listen() {
      return await new Promise((resolve) => {
        server.listen(0, '127.0.0.1', () => {
          const port = server.address().port
          resolve({ port, origin: `http://127.0.0.1:${port}/dshs-relay` })
        })
      })
    },
    close: () => new Promise((resolve) => server.close(() => resolve())),
  }
}

/** 假的"通道句柄"（切流单测不碰真 socket：要测的是**决策**，不是传输）。 */
function fakeChannel(url, health = { state: 'up', attempts: 0, unhealthyForMs: 0 }) {
  const ch = {
    url,
    closed: 0,
    _h: { ...health },
    health: () => ({ ...ch._h }),
    /** 测试用：改成不健康。 */
    setHealth(next) {
      ch._h = { ...next }
    },
    close() {
      ch.closed += 1
    },
  }
  return ch
}

/** 攒日志行（用于断言"判别器可 grep"）。 */
function collector() {
  const lines = []
  return {
    lines,
    log: (l) => lines.push(l),
    /** `[relay-switch]` 开头的行数 —— **必须等于 `switches`**（D7/E9）。 */
    switchLines: () => lines.filter((l) => l.startsWith('[relay-switch]')).length,
  }
}

const sleep = (ms) => new Promise((r) => setTimeout(r, ms))

async function waitFor(fn, timeoutMs = 5000) {
  const deadline = Date.now() + timeoutMs
  for (;;) {
    if (fn()) return true
    if (Date.now() >= deadline) return false
    await sleep(10)
  }
}

/* ─────────── F1 / F2：候选集不再退化成单点（E1 / E2） ─────────── */

test('F1 候选集：目录含两台中继 ⇒ 两条都取出来、顺序 relays[] → bootstrap[]（E1）', async (t) => {
  const keys = makeKeys()
  const dir = serveDirectory({})
  const { port, origin } = await dir.listen()
  t.after(() => dir.close())
  // 目录端点自己也算一个 relay 入口（同源约定）⇒ 用它当"回答目录的那个 origin"
  const A = origin
  const B = `http://127.0.0.1:${port + 1}/dshs-relay`
  const C = `http://127.0.0.1:${port + 2}/dshs-relay`
  Object.assign(dir.state, signedDoc([A, B], keys.privatePem, [C]))

  const opts = {
    seeds: [A],
    trustedKeys: [keys.publicPem],
    cacheFile: '',
    log: () => {},
  }
  const cands = await listOverlayRelayCandidates(opts)
  assert.equal(cands.source, 'seed-directory')
  assert.equal(cands.urls.length, 3, `应列出全部 3 条候选（relays 2 + bootstrap 1），实际 ${JSON.stringify(cands.urls)}`)
  assert.equal(cands.urls[0], 'ws://127.0.0.1:' + port + '/dshs-relay', '首位 = relays[] 第一条（同源优先命中它本身 ⇒ 顺序不变）')
  assert.ok(cands.urls[1].endsWith(`:${port + 1}/dshs-relay`), '第二位 = relays[] 第二条')
  assert.ok(cands.urls[2].endsWith(`:${port + 2}/dshs-relay`), '末位 = bootstrap[]')
})

test('F2 exclude 生效且向后兼容：不传 exclude ⇒ 首位与改造前逐字一致（E2 / D9）', async (t) => {
  const keys = makeKeys()
  const dir = serveDirectory({})
  const { port, origin } = await dir.listen()
  t.after(() => dir.close())
  const A = origin
  const B = `http://127.0.0.1:${port + 1}/dshs-relay`
  Object.assign(dir.state, signedDoc([A, B], keys.privatePem, []))

  const base = { seeds: [A], trustedKeys: [keys.publicPem], cacheFile: '', log: () => {} }
  const plain = await resolveOverlayRelay(base)
  const Aws = `ws://127.0.0.1:${port}/dshs-relay`
  const Bws = `ws://127.0.0.1:${port + 1}/dshs-relay`
  assert.equal(plain.url, Aws, '不传 exclude ⇒ 仍是首位（D9 存量调用点零影响）')

  const excluded = await resolveOverlayRelay({ ...base, exclude: [Aws] })
  assert.equal(excluded.url, Bws, 'exclude 掉当前那台 ⇒ 换到第二台')

  const both = await resolveOverlayRelay({ ...base, exclude: [Aws, Bws] })
  assert.equal(both.url, '', 'D6：候选被排空 ⇒ 返回空串（调用方保持原地退避，⛔ 不切到空）')
  assert.ok(both.detail.endsWith('|exhausted'), '排空时 detail 带 |exhausted 便于取证')
})

/* ─────────── F3：不健康判据可读（E3 / S2 口径） ─────────── */

test('F3 RelayClient 不健康快照：健康 ⇒ 归零；连不上 ⇒ 非空且态为 backoff（E3）', async (t) => {
  // ① 健康：连真的 relay（本文件里唯一一处用真 socket 的地方 —— 要证明"up 状态下确实归零"）
  const secret = randomBytes(32).toString('hex')
  // relay 对声明的端口有两条硬校验：**不能空**（`no-ports`）、**必须在实例口区间内**（`port-out-of-range`）
  // ⇒ 起一个真在监听的 echo 服务，且端口落在 `instancePortBase..+span`
  const BASE = 34400
  const SPAN = 200
  const echo = createTcpServer((sock) => sock.pipe(sock))
  const declared = await new Promise((resolve, reject) => {
    let p = BASE + 7
    const tryBind = () => {
      if (p >= BASE + SPAN) return reject(new Error('no free port in range'))
      echo.once('error', () => {
        p += 1
        tryBind()
      })
      echo.listen(p, '127.0.0.1', () => resolve(echo.address().port))
    }
    tryBind()
  })
  const server = new RelayServer({
    port: 0,
    keys: new Map([['w-f3', secret]]),
    instancePortBase: BASE,
    instancePortSpan: SPAN,
    log: () => {},
  })
  await server.start()
  const ok = new RelayClient({
    url: `ws://127.0.0.1:${server.boundPort}${RELAY_PATH}`,
    hostId: 'w-f3',
    secret,
    ports: [declared],
    log: () => {},
    reconnectMinMs: 10,
    reconnectMaxMs: 40,
  })
  t.after(async () => {
    ok.stop()
    await server.stop()
    echo.close()
  })
  ok.start()
  assert.ok(await waitFor(() => ok.status().state === 'up'), '客户端未在 5s 内 up')
  const healthy = ok.status()
  assert.equal(healthy.unhealthySinceMs, undefined, 'up 状态必须归零（unhealthySinceMs = undefined）')
  assert.equal(healthy.unhealthyForMs, 0, 'up 状态 unhealthyForMs 必须为 0')

  // ② 不健康：指向一个**没人监听**的回环口 ⇒ 必然进 backoff
  const dead = new RelayClient({
    url: 'ws://127.0.0.1:1/dshs-relay',
    hostId: 'w-f3b',
    secret,
    ports: [],
    log: () => {},
    reconnectMinMs: 10,
    reconnectMaxMs: 40,
  })
  t.after(() => dead.stop())
  dead.start()
  assert.ok(await waitFor(() => dead.status().state === 'backoff'), '连不上时必须进入 backoff')
  const bad = dead.status()
  assert.equal(bad.state, 'backoff')
  assert.ok(typeof bad.unhealthySinceMs === 'number', 'backoff 后 unhealthySinceMs 必须非空')
  assert.ok(bad.unhealthyForMs >= 0, 'unhealthyForMs 必须可读（≥ 0）')
})

/* ─────────── F4–F7：切换决策的四条纪律（E4 / E7 / E8 / D4） ─────────── */

/** 搭一个"当前连 A、候选 [A,B]"的监管器。 */
function setup({ candidates = ['A', 'B'], openImpl } = {}) {
  const col = collector()
  const opened = []
  const sup = new RelayFailoverSupervisor({
    open: async (url) => {
      opened.push(url)
      const h = openImpl ? await openImpl(url) : fakeChannel(url)
      return h === undefined ? undefined : h
    },
    candidates: async () => candidates,
    log: col.log,
    thresholds: { minAttempts: 3, graceMs: 1000, cooldownMs: 60_000, deadlineMs: 30_000, checkMs: 10 },
  })
  return { sup, col, opened }
}

test('F4 能切就切：当前不健康 ⇒ 换到下一条，且 switches 与 [relay-switch] 行数相等（D7/E9）', async () => {
  const col = collector()
  const opened = []
  const A = fakeChannel('A', { state: 'backoff', attempts: 3, unhealthyForMs: 1200 })
  const B = fakeChannel('B', { state: 'up', attempts: 0, unhealthyForMs: 0 })
  const sup = new RelayFailoverSupervisor({
    open: async (url) => {
      opened.push(url)
      return url === 'B' ? B : undefined
    },
    candidates: async () => ['A', 'B'],
    log: col.log,
    thresholds: { minAttempts: 3, graceMs: 1000, cooldownMs: 60_000, deadlineMs: 30_000, checkMs: 10 },
  })
  sup.seed(A)
  await sup.tick()

  const st = sup.stats()
  assert.deepEqual(opened, ['B'], '只应去建 B 这一条')
  assert.equal(st.switches, 1, '必须切换一次')
  assert.equal(sup.channel, B, '当前通道必须已是 B')
  assert.equal(A.closed, 1, '旧通道必须被关掉（先建新、成功再关旧）')
  assert.equal(B.closed, 0, '新通道不能被关')
  assert.equal(col.switchLines(), st.switches, 'D7：日志行数必须等于 switches 计数')
  assert.match(col.lines.find((l) => l.startsWith('[relay-switch]')), /#1 A -> B/)
  assert.equal(st.cooldown.length, 1, '被换掉的 A 必须进冷却表')
  assert.equal(st.cooldown[0].url, 'A')
})

test('F5 无候选不切空：链里只剩当前那台 ⇒ 原地退避、switches 不增、不静默回退（D6/E7）', async () => {
  const { sup, col } = setup({ candidates: ['A'] })
  const A = fakeChannel('A', { state: 'backoff', attempts: 8, unhealthyForMs: 9000 })
  sup.seed(A)
  await sup.tick()
  await sup.tick()

  const st = sup.stats()
  assert.equal(st.switches, 0, '⛔ 不许切到空')
  assert.equal(sup.channel, A, '⛔ 不许静默回退到别的机器')
  assert.ok(st.noCandidateChecks >= 1, '必须记下"无候选可切"的次数（D6 现场证据）')
  assert.equal(col.switchLines(), 0, '没有切换 ⇒ 不许有 [relay-switch] 行')
  assert.ok(col.lines.some((l) => l.startsWith('[relay-skip]')), '必须有一行 [relay-skip] 说明为何不切')
})

/**
 * 🔴 **序⑧ 的冲突与裁决（F6 / F9 为什么多了 `exempt: false`）**：
 *
 * 本用例的场景**恰好就是**序⑧ 要改的那个现场 —— 当前通道 `B` 不健康、链里唯一的替代 `A` 正在冷却
 * ⇒ `tick()` 过滤后 `target === undefined` = **D6 现场**。序⑧ 的 D1 在这里**故意**开了"一跳豁免"
 * （D1 原文："旧 url 已被证伪 ⇒ 回跳它不是抖动，而是**唯一可能的出路**"）。
 * 而交接单 §3.1-5 / E4 又要求"序⑦ F1–F11 全绿、不许改语义" —— 两者在**这个场景**上真冲突
 * （真机 E9 幕 4 与它同构 ⇒ 没有任何判据能"只豁免幕 4、不豁免 F6"）。
 *
 * 裁决（依 D3 的**精确口径**）：D3 给的护栏是"**有干净候选时**行为逐字不变" ——
 * 本场景**没有**干净候选，所以不在 D3 的保证范围内。⇒
 * **断言逐字不变**，只把"关闭序⑧ 豁免"这个开关显式写进用例配置 ⇒
 * 本用例继续锁住 **D5 的基线语义**（= 序⑦ 逐字行为），序⑧ 的新行为由 **F13 / F15 / F16 / F17** 锁住。
 */
test('F6 冷却期内不回跳：A 恢复了但仍在冷却 ⇒ 不换回 A（D5/E8；豁免关 = 序⑦ 基线）', async () => {
  const col = collector()
  let A
  const sup = new RelayFailoverSupervisor({
    open: async (url) => (url === 'B' ? fakeChannel('B') : fakeChannel(url)),
    candidates: async () => ['A', 'B'],
    log: col.log,
    thresholds: { minAttempts: 3, graceMs: 1000, cooldownMs: 60_000, deadlineMs: 30_000, checkMs: 10, exempt: false },
  })
  A = fakeChannel('A', { state: 'backoff', attempts: 5, unhealthyForMs: 5000 })
  sup.seed(A)
  await sup.tick()
  assert.equal(sup.stats().switches, 1, '第一次：A 挂 ⇒ 切 B')

  // B（当前）也变成不健康；此时 A 已"恢复"（健康）但**在冷却期内**
  const B = sup.channel
  B.setHealth({ state: 'backoff', attempts: 5, unhealthyForMs: 5000 })
  A.setHealth({ state: 'up', attempts: 0, unhealthyForMs: 0 })
  await sup.tick()

  assert.equal(sup.stats().switches, 1, '⛔ 冷却期内不回跳（否则两台互相抢 = 抖动风暴）')
  assert.equal(sup.channel, B, '仍应留在 B')
})

test('F7 新通道建不起来 ⇒ 原通道原样保留（D4 / R11）', async () => {
  const { sup, col } = setup({ openImpl: async () => undefined })
  const A = fakeChannel('A', { state: 'backoff', attempts: 9, unhealthyForMs: 9000 })
  sup.seed(A)
  await sup.tick()

  const st = sup.stats()
  assert.equal(st.switches, 0, '建不起来就不算切换')
  assert.equal(st.openFailed, 1, '必须记下 openFailed')
  assert.equal(sup.channel, A, '⛔ 原通道必须保留（把本来能用的通路打掉才是真事故）')
  assert.equal(A.closed, 0, '⛔ 不许把旧通道关掉')
  assert.equal(col.switchLines(), 0)
})

test('F8 巡检只在"不健康"时动手：健康通道连跑多轮 ⇒ 零切换、零 open', async () => {
  const { sup, opened, col } = setup()
  const A = fakeChannel('A', { state: 'up', attempts: 0, unhealthyForMs: 0 })
  sup.seed(A)
  for (let i = 0; i < 5; i += 1) await sup.tick()
  assert.equal(opened.length, 0, '健康时不许调 open（否则每次巡检都白建一条通道）')
  assert.equal(sup.stats().switches, 0)
  assert.equal(col.switchLines(), 0)
})

/* ─────────── F9 / F10：真机拓扑逼出来的两条（注入时钟 / 死候选） ─────────── */

/**
 * F9 · **冷却期满 ⇒ 自动回归候选表**（D5 后半句）。
 *
 * 为什么必须用注入时钟：真机冷却 300 s，等不起。等不起的判据就等于没有 ⇒ 必须可确定性复现。
 *
 * ⚠️ 序⑧：本用例中段（"冷却未满 ⇒ 不回跳"）同样是 **D6 现场** ⇒ 与 F6 同因，显式置
 * `exempt: false`（= 锁序⑦ 基线；序⑧ 的新行为见 F13/F15/F16/F17）。
 */
test('F9 冷却期满 ⇒ 失败过的那台自动回归候选表（D5；豁免关 = 序⑦ 基线）', async () => {
  const col = collector()
  let now = 1_000_000
  const sup = new RelayFailoverSupervisor({
    open: async (url) => fakeChannel(url),
    candidates: async () => ['A', 'B'],
    log: col.log,
    nowMs: () => now,
    thresholds: { minAttempts: 3, graceMs: 1000, cooldownMs: 60_000, deadlineMs: 30_000, checkMs: 10, exempt: false },
  })
  const A = fakeChannel('A', { state: 'backoff', attempts: 5, unhealthyForMs: 5000 })
  sup.seed(A)
  await sup.tick()
  assert.equal(sup.stats().switches, 1, 'A 挂 ⇒ 切 B，A 进冷却')
  assert.equal(sup.channel.url, 'B')

  // B 也挂；此刻 A 已"恢复"，但**冷却未满** ⇒ 不回跳
  sup.channel.setHealth({ state: 'backoff', attempts: 5, unhealthyForMs: 5000 })
  A.setHealth({ state: 'up', attempts: 0, unhealthyForMs: 0 })
  now += 59_000
  await sup.tick()
  assert.equal(sup.stats().switches, 1, '冷却未满 ⇒ 不回跳')

  // 冷却期满 ⇒ A 回归候选 ⇒ 允许换回 A
  now += 2_000
  await sup.tick()
  assert.equal(sup.stats().switches, 2, '冷却期满 ⇒ 回归候选并换回')
  assert.equal(sup.channel.url, 'A')
  assert.equal(col.switchLines(), 2, 'D7：日志行数仍等于 switches')
})

/**
 * F10 · **试失败的候选不能把链堵死**（真机拓扑逼出来的一条）。
 *
 * 生产目录 `relays[]` 前两条**落在同一台机器**上：杀那台时，若失败的候选不进冷却，
 * 每次巡检都会卡在同一条上、**永远推进不到第三条** ⇒ 链"不再退化成单点"却依然换不过去。
 */
test('F10 前两个候选建不起来 ⇒ 自动推进到第三个（失败候选进冷却，⛔ 不堵链）', async () => {
  const col = collector()
  const alive = fakeChannel('C', { state: 'up', attempts: 0, unhealthyForMs: 0 })
  const tried = []
  const sup = new RelayFailoverSupervisor({
    open: async (url) => {
      tried.push(url)
      return url === 'C' ? alive : undefined // A / B 同机已死
    },
    candidates: async () => ['A', 'B', 'C'],
    log: col.log,
    thresholds: { minAttempts: 3, graceMs: 1000, cooldownMs: 60_000, deadlineMs: 30_000, checkMs: 10 },
  })
  const A = fakeChannel('A', { state: 'backoff', attempts: 5, unhealthyForMs: 5000 })
  sup.seed(A)

  await sup.tick() // 试 B（A 是当前 ⇒ 被排除）⇒ 失败 ⇒ 进冷却
  assert.equal(sup.stats().switches, 0, '第一次不该算切换')
  await sup.tick() // B 已在冷却 ⇒ 推进到 C ⇒ 成功
  assert.equal(sup.stats().switches, 1, '第二次必须推进到 C 并切换成功')
  assert.equal(sup.channel, alive)
  assert.deepEqual(tried, ['B', 'C'], '尝试顺序必须是"下一个候选 → 再下一个"（⛔ 不许卡在 B 上反复试）')
  assert.equal(col.switchLines(), sup.stats().switches)
})

/**
 * F11 · **冷却闸门对"目录地址变更"这条路径同样生效**（D5）。
 *
 * 真机实测逼出来的一条：`refreshOverlay`（目录地址变了）直接调 `replace()`，
 * 它**不看冷却表** ⇒ 会把刚被冷却的地址立刻换回来 ⇒ 抖动抑制被绕开。
 */
test('F11 冷却期内的目标：连"目录地址变更"路径也不许换过去（D5）', async () => {
  const col = collector()
  let now = 5_000_000
  const sup = new RelayFailoverSupervisor({
    open: async (url) => fakeChannel(url),
    candidates: async () => ['A', 'B'],
    log: col.log,
    nowMs: () => now,
    thresholds: { minAttempts: 3, graceMs: 1000, cooldownMs: 60_000, deadlineMs: 30_000, checkMs: 10 },
  })
  const A = fakeChannel('A', { state: 'backoff', attempts: 5, unhealthyForMs: 5000 })
  sup.seed(A)
  await sup.tick() // A 挂 ⇒ 切 B，A 进冷却
  assert.equal(sup.channel.url, 'B')
  assert.equal(sup.stats().switches, 1)

  // 模拟 `refreshOverlay`：目录首位又变回 A（刚被冷却）
  // ⚠️ 序⑧ 起 `replace()` 必须显式声明 `origin`（'directory' = **无豁免权**，D1）——
  //    本用例的语义与序⑦ 逐字相同（仍是"冷却期内不许换过去"），只是把隐含意图变成显式参数。
  const ok = await sup.replace('A', '目录地址变更（source=cache）', 'directory')
  assert.equal(ok, false, '⛔ 冷却期内的目标不许换过去（否则抖动抑制形同不存在）')
  assert.equal(sup.stats().switches, 1, 'switches 不许增加')
  assert.equal(sup.channel.url, 'B', '仍应留在 B')

  // 冷却期满 ⇒ 允许换回 A
  now += 61_000
  assert.equal(await sup.replace('A', '目录地址变更（source=cache）', 'directory'), true, '冷却期满 ⇒ 允许')
  assert.equal(sup.channel.url, 'A')
})

/* ─────────── F12–F17：序⑧「切流冷却语义」（D1–D6） ─────────── */

/** 序⑧ 的公共阈值：与 F9/F11 同口径（可注入时钟 ⇒ 冷却期满可确定性复现）。 */
const TH8 = { minAttempts: 3, graceMs: 1000, cooldownMs: 60_000, deadlineMs: 30_000, checkMs: 10 }

/**
 * F12 · **目录路径没有豁免权**（E1 / D1）。
 *
 * 立项依据：真机 11:43:26 实测 `wss://106… -> wss://example.net…` —— 只因"目录里的地址变了"
 * 就把刚被冷却的 47 换回来 ⇒ D5 的抖动抑制被另一条路径绕开。
 */
test('F12 目录路径无豁免权：origin=directory ＋ 目标在冷却 ⇒ 必 skip（E1 / D1）', async () => {
  const col = collector()
  let now = 1_000_000
  const sup = new RelayFailoverSupervisor({
    open: async (url) => fakeChannel(url),
    candidates: async () => ['A', 'B'],
    log: col.log,
    nowMs: () => now,
    thresholds: TH8,
  })
  const A = fakeChannel('A', { state: 'backoff', attempts: 5, unhealthyForMs: 5000 })
  sup.seed(A)
  await sup.tick() // A 挂 ⇒ 切 B；A 进冷却（kind = switched-away）
  assert.equal(sup.channel.url, 'B')
  assert.equal(sup.stats().switches, 1)
  assert.deepEqual(
    sup.stats().cooldown.map((c) => [c.url, c.kind]),
    [['A', 'switched-away']],
    '冷却表已结构化：键仍按 url，`kind` 记录"我们主动离开了它"',
  )

  now += 5_000
  const ok = await sup.replace('A', '目录地址变更（source=cache）', 'directory')
  assert.equal(ok, false, '⛔ 目录路径**不得**打破冷却（D1）')
  assert.equal(sup.stats().switches, 1, 'switches 不许增加')
  assert.equal(sup.stats().exemptSwitches, 0, '⛔ 这不是豁免')
  assert.equal(sup.channel.url, 'B', '仍应留在 B')
  assert.ok(
    col.lines.some((l) => l.startsWith('[relay-skip]') && l.includes('仍在冷却')),
    '必须留下"仍在冷却"的判别器行',
  )
  assert.ok(
    col.lines.every((l) => !l.includes('｜豁免')),
    '⛔ 目录路径不许出现豁免标记',
  )
})

/**
 * F13 · **一跳豁免成立**（E2 / D1 / D4）。
 *
 * 现场 = D6：生产目录 3 条候选里 **2 条同机** ⇒ 一次 47 故障把它们**同时**耗进冷却 ⇒
 * "当前这条也挂了"时链里再无干净候选 ⇒ 序⑧ 之前是**最长 `cooldownMs` 不切流**。
 */
test('F13 一跳豁免：D6 现场 ＋ 1 条 switched-away ⇒ 切过去并计数（E2 / D4）', async () => {
  const col = collector()
  let now = 2_000_000
  const sup = new RelayFailoverSupervisor({
    open: async (url) => fakeChannel(url),
    candidates: async () => ['A', 'B'],
    log: col.log,
    nowMs: () => now,
    thresholds: TH8,
  })
  const A = fakeChannel('A', { state: 'backoff', attempts: 5, unhealthyForMs: 5000 })
  sup.seed(A)
  await sup.tick() // A 挂 ⇒ 切 B；A 进冷却
  assert.equal(sup.channel.url, 'B')

  // B（当前）也挂；此刻 A **仍在冷却窗内**（60 s）⇒ 候选池被耗干 = D6 现场
  sup.channel.setHealth({ state: 'backoff', attempts: 5, unhealthyForMs: 5000 })
  A.setHealth({ state: 'up', attempts: 0, unhealthyForMs: 0 })
  now += 5_000
  await sup.tick()

  const st = sup.stats()
  assert.equal(st.switches, 2, '豁免必须完成一次真实切换')
  assert.equal(st.exemptSwitches, 1, 'D7：豁免切换必须计数（⊆ switches）')
  assert.equal(sup.channel.url, 'A', '必须切回 A')
  assert.equal(st.noCandidateChecks, 0, '豁免成功 ⇒ 不该再记"无候选"')
  const sw = col.lines.filter((l) => l.startsWith('[relay-switch]'))
  assert.equal(sw.length, st.switches, 'D7：日志行数必须等于 switches（豁免行也是 switch 行）')
  assert.match(sw[1], /｜豁免 kind=switched-away /, '豁免切换必须带可断言的判别器标记')
  assert.match(sw[1], /原因：当前通道不健康/, 'E9③：原因必须是 health 路径，⛔ 不是"目录地址变更"')
})

/**
 * F14 · **豁免优先级**（E3 / D5）：`switched-away` 优先于 `open-failed`。
 *
 * 语义依据：`switched-away` = "我们主动离开了一件**曾可用**的东西"；`open-failed` = "刚证明它建不起来"。
 * 本用例里 `open-failed` 的 `untilMs` **更早**（先失败先解除），仍然必须让位于 `switched-away`
 * ⇒ 同时验证"优先级**压过** `untilMs` 升序"。
 */
test('F14 豁免优先级：switched-away 压过 open-failed（E3 / D5）', async () => {
  const col = collector()
  let now = 3_000_000
  const tried = []
  const sup = new RelayFailoverSupervisor({
    open: async (url) => {
      tried.push(url)
      return url === 'X' ? undefined : fakeChannel(url) // X 永远建不起来
    },
    candidates: async () => ['C', 'X', 'Y'],
    log: col.log,
    nowMs: () => now,
    thresholds: TH8,
  })
  const C = fakeChannel('C', { state: 'backoff', attempts: 5, unhealthyForMs: 5000 })
  sup.seed(C)
  await sup.tick() // 试 X ⇒ 失败 ⇒ X 进冷却（open-failed）
  assert.equal(sup.stats().openFailed, 1)
  now += 1_000
  await sup.tick() // 推进到 Y ⇒ 成功；C 进冷却（switched-away）
  assert.equal(sup.channel.url, 'Y')
  assert.deepEqual(
    sup.stats().cooldown.map((c) => [c.url, c.kind]),
    [
      ['X', 'open-failed'],
      ['C', 'switched-away'],
    ],
    '两条冷却条目：X 先建冷却（untilMs 更早），C 后建',
  )

  sup.channel.setHealth({ state: 'backoff', attempts: 5, unhealthyForMs: 5000 })
  now += 1_000
  await sup.tick() // D6 现场 ⇒ 豁免：必须挑 C（switched-away），⛔ 不是 untilMs 更早的 X

  assert.deepEqual(tried.slice(-1), ['C'], '豁免必须挑 switched-away 那条（压过 untilMs 升序）')
  assert.equal(sup.channel.url, 'C')
  assert.equal(sup.stats().exemptSwitches, 1)
})

/**
 * F15 · 🔴 **D3 护栏：有干净候选时行为逐字不变**。
 *
 * 这是本单**最重要的一条不变量** —— 豁免一旦泄漏进正常路径，就会变成"每轮巡检都想回跳"（新抖动源），
 * 等于把序⑦ 的 E1–E11 结论**自己推翻自己**。
 */
test('F15 有干净候选 ⇒ 绝不走豁免（D3 护栏）', async () => {
  const col = collector()
  let now = 4_000_000
  const sup = new RelayFailoverSupervisor({
    open: async (url) => (url === 'B' ? undefined : fakeChannel(url)), // B 永远建不起来
    candidates: async () => ['A', 'B', 'C', 'D'],
    log: col.log,
    nowMs: () => now,
    thresholds: TH8,
  })
  const A = fakeChannel('A', { state: 'backoff', attempts: 5, unhealthyForMs: 5000 })
  sup.seed(A)
  await sup.tick() // 试 B ⇒ 失败 ⇒ B 进冷却
  now += 1_000
  await sup.tick() // 推进到 C ⇒ 成功；A 进冷却
  assert.equal(sup.channel.url, 'C')
  assert.equal(sup.stats().switches, 1)

  // 当前 C 也挂：此时 A（switched-away）/ B（open-failed）都在冷却，但 **D 是干净的**
  sup.channel.setHealth({ state: 'backoff', attempts: 5, unhealthyForMs: 5000 })
  A.setHealth({ state: 'up', attempts: 0, unhealthyForMs: 0 })
  now += 1_000
  await sup.tick()

  const st = sup.stats()
  assert.equal(sup.channel.url, 'D', '有干净候选 ⇒ 必须走**正常**候选链（D3）')
  assert.equal(st.switches, 2)
  assert.equal(st.exemptSwitches, 0, '⛔ 豁免一次都不许发生')
  assert.equal(st.noCandidateChecks, 0, '⛔ 也不该记"无候选"')
  assert.ok(
    col.lines.every((l) => !l.includes('｜豁免')),
    '⛔ 日志里不许出现豁免标记（逐字回到序⑦）',
  )
})

/**
 * F16 · **豁免有界**（E5 / D4）：每 url **每冷却周期一次**；豁免再失败 ⇒ 重置冷却且本周期不再豁免。
 *
 * ⛔ 不设界 = "每 2 s 豁免一次、每次都失败" = **重试风暴**，比不切更糟（R11）。
 * ⚠️ 全程**不推进注入时钟** ⇒ 证明"同一冷却周期内"。两轮豁免各失败一次会让 `openFailed` 递增，
 * 但**同一个 url 只会被试一次**。
 */
test('F16 豁免有界：同周期内每个冷却候选只豁免一次、不再重复试（E5 / D4）', async () => {
  const col = collector()
  let now = 6_000_000
  const tried = []
  const sup = new RelayFailoverSupervisor({
    open: async (url) => {
      tried.push(url)
      return url === 'C' ? fakeChannel('C') : undefined // 只有 C 能起
    },
    candidates: async () => ['A', 'B', 'C'],
    log: col.log,
    nowMs: () => now,
    thresholds: TH8,
  })
  const A = fakeChannel('A', { state: 'backoff', attempts: 5, unhealthyForMs: 5000 })
  sup.seed(A)
  await sup.tick() // 试 B ⇒ 失败 ⇒ B 进冷却（open-failed）
  await sup.tick() // 推进到 C ⇒ 成功；A 进冷却（switched-away）
  assert.equal(sup.channel.url, 'C')
  assert.equal(sup.stats().switches, 1)
  assert.equal(sup.stats().openFailed, 1)

  // C 也挂 ⇒ D6 现场（A、B 均冷却）。此后**不推进时钟** ⇒ 全程同一冷却周期。
  sup.channel.setHealth({ state: 'backoff', attempts: 5, unhealthyForMs: 5000 })
  await sup.tick() // 豁免 A（switched-away 优先）⇒ 失败 ⇒ A 本周期额度用尽
  assert.equal(sup.stats().openFailed, 2, '第一次豁免失败必须计数')
  await sup.tick() // A 已被排除 ⇒ 豁免 B ⇒ 也失败
  assert.equal(sup.stats().openFailed, 3, '第二个冷却候选仍可豁免一次')
  await sup.tick() // 池子空了 ⇒ 回到原地退避，⛔ 不许再试 A/B
  await sup.tick()

  const st = sup.stats()
  assert.equal(st.openFailed, 3, '⛔ 重试风暴：同一 url 每周期只许试一次')
  assert.equal(st.switches, 1, '豁免全失败 ⇒ 不许增加 switches')
  assert.equal(st.exemptSwitches, 0)
  assert.ok(st.noCandidateChecks >= 2, '池子空了必须记"无候选"（D6 现场证据）')
  assert.deepEqual(tried, ['B', 'C', 'A', 'B'], '尝试序列：B（正常）→ C（正常）→ A（豁免）→ B（豁免）')
  assert.ok(
    col.lines.some((l) => l.includes('本周期不再豁免')),
    '必须留下"本周期不再豁免"的判别器行',
  )
  assert.equal(col.switchLines(), st.switches, 'D7：行数仍等于 switches')
})

/**
 * F17 · **两层开关各司其职**（E6 / D6）：`RELAY_FAILOVER_EXEMPT=0` ⇒ 逐字回到序⑦ 行为。
 */
test('F17 总开关 RELAY_FAILOVER_EXEMPT=0 ⇒ 无豁免（第二层回滚点）', async () => {
  const col = collector()
  let now = 7_000_000
  const sup = new RelayFailoverSupervisor({
    open: async (url) => fakeChannel(url),
    candidates: async () => ['A', 'B'],
    log: col.log,
    nowMs: () => now,
    thresholds: { ...TH8, exempt: false },
  })
  const A = fakeChannel('A', { state: 'backoff', attempts: 5, unhealthyForMs: 5000 })
  sup.seed(A)
  await sup.tick() // A 挂 ⇒ 切 B；A 进冷却
  assert.equal(sup.channel.url, 'B')

  sup.channel.setHealth({ state: 'backoff', attempts: 5, unhealthyForMs: 5000 })
  A.setHealth({ state: 'up', attempts: 0, unhealthyForMs: 0 })
  now += 5_000
  await sup.tick()

  const st = sup.stats()
  assert.equal(st.switches, 1, '⛔ 开关关闭 ⇒ 回到"原地退避"（序⑦ 行为，逐字）')
  assert.equal(st.exemptSwitches, 0)
  assert.equal(sup.channel.url, 'B')
  assert.ok(st.noCandidateChecks >= 1, '必须回到"链里无其他候选"的原地退避路径')
  assert.ok(col.lines.every((l) => !l.includes('｜豁免')), '⛔ 日志里不许出现豁免标记')
})

/* ═══════════════════════════════════════════════════════════════════════════
 * 序⑨ · **换址等待的"终态失败"**（RC-1）—— F18–F21
 *
 * 背景（实测，序⑨ §2-P10 逐行复核）：`waitUpOn` 只轮询 `state === 'up'` ⇒ **死候选与慢候选
 * 不可区分**，代价恒为 `upTimeoutMs`(12 s)。生产目录前两条候选**同在 47** ⇒ 每次从 47 切走
 * 都先试同机的 `relay-direct`（已随 47 一起死）⇒ **固定白等 12 s**，占 30 s 墙钟的 40%。
 *
 * 纪律：① 修（F19）必须有**护栏**（F18：慢候选不许被误杀）② burst 窗口（计划内重启）
 * **不许**被当终态失败（F20，否则每次 relay 重启都切流 = D3 要防的抖动）。
 * ═══════════════════════════════════════════════════════════════════════════ */

/**
 * 假客户端：按脚本在给定毫秒数后切换 `status()`（⛔ 不碰网络 —— 本组只验"等待逻辑"这一层）。
 * `plan` = `[{ at: 毫秒（相对本次调用开始）, st: 状态片段 }]`，后者覆盖前者。
 */
function fakeWaitee(plan, base = {}) {
  const t0 = Date.now()
  const seen = []
  return {
    seen,
    status() {
      const el = Date.now() - t0
      let st = { state: 'connecting', attempts: 0, inGracefulBurstWindow: false, ...base }
      for (const p of plan) if (el >= p.at) st = { ...st, ...p.st }
      seen.push(st.state)
      return st
    },
  }
}

/**
 * F21 · 判据的**切面**（先于行为用例：把"什么算终态失败"钉死在四个反例上）。
 */
test('F21 终态失败判据的四个反例：connecting / handshaking / queued(attempts=0) / burst 窗口内 ⇒ 都不算', () => {
  const dead = { state: 'backoff', attempts: 1, inGracefulBurstWindow: false }
  assert.equal(openedChannelFailedTerminally(dead), true, 'backoff ＋ 已记账 ＋ 非 burst ⇒ 终态失败')

  const cases = [
    ['正在连（connecting）', { ...dead, state: 'connecting' }],
    ['正在握手（handshaking）', { ...dead, state: 'handshaking' }],
    ['满载排队（queued ⇒ attempts 被归零）', { ...dead, attempts: 0 }],
    ['计划内重启的 burst 窗口内', { ...dead, inGracefulBurstWindow: true }],
  ]
  for (const [name, st] of cases) {
    assert.equal(openedChannelFailedTerminally(st), false, `${name} ⛔ 不许判成终态失败`)
  }
})

/**
 * F19 · **死候选 ⇒ 提前失败**（本单的核心修法；⛔ 这是判据的"红→绿"分水岭）。
 * 旧实现下本断言必然失败：白等满 12 000 ms。
 */
test('F19 死候选：`backoff` ＋ 已记账 ＋ 非 burst ⇒ 提前失败（⛔ 不白等满 upTimeoutMs）', async () => {
  const dead = fakeWaitee([
    { at: 100, st: { state: 'backoff', attempts: 1, inGracefulBurstWindow: false, lastError: 'transport error' } },
  ])
  const t0 = Date.now()
  const ok = await waitUpOnStatus(dead, 12_000)
  const ms = Date.now() - t0
  assert.equal(ok, false, '死候选必须返回 false')
  assert.ok(ms < 2_000, `必须远早于 upTimeoutMs(12000) 返回；旧实现会白等满，实测 ${ms}ms`)
})

/**
 * F18 · **护栏：慢候选不许被误杀**（R11 不变量）。
 *
 * 慢候选在到达 `up` 之前**一直处于 `connecting`**，从不进 `backoff` ⇒ 判据恒不成立 ⇒
 * 照旧享受完整 `upTimeoutMs`。⛔ 这条是"早退"的安全带：没有它，早退会退化成
 * "更频繁地切到第三候选"甚至"全部候选都判失败"。
 */
test('F18 护栏：慢候选（连得上、`up` 来得晚）⇒ 仍等满预算且必须成功', async () => {
  const slow = fakeWaitee([{ at: 800, st: { state: 'up', attempts: 0 } }])
  const t0 = Date.now()
  const ok = await waitUpOnStatus(slow, 12_000)
  const ms = Date.now() - t0
  assert.equal(ok, true, '慢候选必须成功 —— ⛔ 早退不许误杀')
  assert.ok(ms >= 600 && ms < 5_000, `应等到 ~800ms 它自己 up 为止，实测 ${ms}ms`)
})

/**
 * F22 · **`gracefulBurstMs` 参数表化**（序⑨ §3.1-4）：默认值语义**逐字不变**，只是可配了。
 * ⛔ 只加可配性、⛔ 不改默认值（E10 的同族纪律：判据/阈值不许被悄悄放宽）。
 */
test('F22 RELAY_GRACEFUL_BURST_MS：默认 15000 逐字不变，显式覆写才生效', () => {
  assert.equal(gracefulBurstMsDefault({}), 15_000, '⛔ 默认值必须逐字不变')
  assert.equal(gracefulBurstMsDefault({ RELAY_GRACEFUL_BURST_MS: '' }), 15_000, '空串 ⇒ 回落默认')
  assert.equal(gracefulBurstMsDefault({ RELAY_GRACEFUL_BURST_MS: 'abc' }), 15_000, '非法值 ⇒ 回落默认（⛔ 不抛）')
  assert.equal(gracefulBurstMsDefault({ RELAY_GRACEFUL_BURST_MS: '-1' }), 15_000, '负数 ⇒ 回落默认')
  assert.equal(gracefulBurstMsDefault({ RELAY_GRACEFUL_BURST_MS: '0' }), 0, '0 = 显式关掉 burst 窗口（合法值）')
  assert.equal(gracefulBurstMsDefault({ RELAY_GRACEFUL_BURST_MS: '30000' }), 30_000, '显式覆写生效')
})

/**
 * F20 · **计划内重启（burst 窗口）⛔ 不算终态失败**（D3 保护）。
 *
 * 窗口内 `state=backoff attempts=1` 与"死候选"**字面完全一样**，唯一区分依据就是
 * `inGracefulBurstWindow`。⛔ 若把它当死候选 ⇒ **每次 relay 重启 / 部署都切一次流**。
 */
test('F20 burst 窗口（计划内重启）内必须继续等，窗口过后才允许判死（D3 保护）', async () => {
  const restarting = fakeWaitee([
    { at: 100, st: { state: 'backoff', attempts: 1, inGracefulBurstWindow: true } },
    { at: 1_500, st: { state: 'up', attempts: 0 } },
  ])
  const t0 = Date.now()
  const ok = await waitUpOnStatus(restarting, 12_000)
  const ms = Date.now() - t0
  assert.equal(ok, true, 'burst 窗口内必须继续等 ⇒ 对端重启完就 up')
  assert.ok(ms >= 1_300, `⛔ 不许在窗口内就判死（旧坑：100ms 处就返回 false）；实测 ${ms}ms`)
})

/* ─────────── 序㉗：候选链观测（`OBS-21`「每连接候选数 ≥ 2」的判据锚点） ─────────── */

/**
 * O1 · **观测行的固定 key 序就是探针的判据锚点** ⇒ 逐字锁住。
 *
 * 🔴 为什么这条最要紧：探针 `OBS-21` 是按 `key=value` **按名取值**的 ⇒ 谁把键改名 / 把值里的
 * 空白留在行里 / 少写一个键，探针会**静默取不到**（本线最贵的一类失效：不是报错，是"看不见"）。
 */
test('O1 观测行格式锁定：固定 key 序 ＋ count 为条数 ＋ hosts 按主机去重（丢 scheme）', () => {
  const lines = []
  const obs = new RelayCandidateObservation('manager', (l) => lines.push(l), 0)
  obs.record(
    [
      'wss://example.net/dshs-relay',
      'https://example.net/other',
      'wss://198.51.100.20/dshs-relay',
    ],
    'cache',
    '/var/lib/dsh-test/overlay/directory.json',
  )
  assert.equal(lines.length, 1, '一次 record 写一行')
  const line = lines[0]
  assert.ok(line.startsWith(CAND_OBS_PREFIX), `必须以固定前缀开头：${line}`)
  const keys = [...line.matchAll(/(?:^|\s)([a-z]+)=/g)].map((m) => m[1])
  assert.deepEqual(
    keys,
    ['scope', 'resolves', 'count', 'hosts', 'source', 'detail', 'urls'],
    '固定 key 序 = 契约（⛔ 改它 = 破坏探针判据）',
  )
  const snap = obs.snapshot()
  assert.equal(snap.count, 3, 'count = 候选**条数**（⛔ 不按主机去重）')
  assert.equal(snap.hosts, 2, 'hosts = 独立主机数（example.net 的两条算同一台 —— scheme 不参与）')
  assert.equal(snap.source, 'cache')
  assert.equal(snap.resolves, 1)
  assert.equal(snap.unresolved, false)
})

/**
 * O2 · **稳态不刷屏**：`RELAY_FAILOVER_CHECK_MS` 是 2 s，同形状会被反复解析 ⇒ 变化才写。
 * ⚠️ 但解析次数必须**照实累计**（探针拿 `resolves` 判"这个进程到底解析过没有"）。
 */
test('O2 同形状重复 record ⛔ 不重复写行（防刷屏），形状一变立刻写', () => {
  const lines = []
  const obs = new RelayCandidateObservation('worker', (l) => lines.push(l), 0)
  const urls = ['wss://a.example/dshs-relay']
  obs.record(urls, 'chain', '')
  obs.record(urls, 'chain', '')
  obs.record(urls, 'chain', '')
  assert.equal(lines.length, 1, '稳态巡检 ⛔ 不许刷屏')
  assert.equal(obs.snapshot().resolves, 3, '解析次数必须照实累计')
  obs.record(['wss://a.example/dshs-relay', 'wss://b.example/dshs-relay'], 'chain', '')
  assert.equal(lines.length, 2, '条数变了（候选集变化）⇒ 必须立刻写')
  assert.equal(obs.snapshot().count, 2)
})

/**
 * O3 · 🔴 **「从未解析」与「解析出 0 条」必须可区分**（本线两处静默失效都栽在这一点），
 * 且**周期重发在零网络下也能写出行**（探针是**事后**读，没有它就可能读不到行）。
 */
test('O3 「从未解析」≠「解析出 0 条」＋ 周期重发零网络写行 ＋ stop 后不再写', async () => {
  const lines = []
  const obs = new RelayCandidateObservation('worker', (l) => lines.push(l), 5)
  const s0 = obs.snapshot()
  assert.equal(s0.unresolved, true, '没解析过 ⇒ unresolved')
  assert.equal(s0.source, 'unresolved')
  assert.equal(s0.resolves, 0)
  obs.start()
  await new Promise((r) => setTimeout(r, 40))
  obs.stop()
  assert.ok(lines.length >= 2, `周期重发必须写出行（实测 ${lines.length} 行）`)
  assert.ok(
    lines.every((l) => l.includes('source=unresolved')),
    '未解析时重发的行也必须**诚实**写 unresolved（⛔ 不许假装 0 条 = 已解析）',
  )
  const n = lines.length
  await new Promise((r) => setTimeout(r, 30))
  assert.equal(lines.length, n, 'stop() 后 ⛔ 不许再写')
  obs.record([], 'none', 'none')
  assert.equal(obs.snapshot().unresolved, false, '解析过就是解析过 —— 哪怕解析出 0 条')
  assert.equal(obs.snapshot().count, 0)
  assert.equal(obs.snapshot().hosts, 0)
})

/** O4 · 重发周期取自 env（与 `switcher.ts#relayFailoverThresholds` 同纪律：值格必须纯数字）。 */
test('O4 重发周期取自 env，`0` = 关闭，非纯数字 ⇒ 回退默认（不静默变成 NaN）', () => {
  assert.equal(candidateObsMs({}), 300_000, '缺省 300 s')
  assert.equal(candidateObsMs({ RELAY_CAND_OBS_MS: '0' }), 0, '0 = 关闭周期重发')
  assert.equal(candidateObsMs({ RELAY_CAND_OBS_MS: '60000' }), 60_000, '显式覆写生效')
  assert.equal(candidateObsMs({ RELAY_CAND_OBS_MS: '6e4' }), 300_000, '非纯数字 ⇒ 回退默认')
})