fix: gate watchdog recovery on bounded runtime observation

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## 未发布
 
+- 内核看门狗在容器级恢复/重建前先对槽位所在 Docker 主机做有界观测（运行/缺失/已停/Restarting/暂停/unknown）；HTTP/权限/传输失败或观测超时视为 unknown，不消耗重启预算、不盲重启；确认 missing/stopped/restarting 且槽位仍为运行意图时才 revival/recreate，运行中容器仍走既有 kernel 重启；远程 Restarting 崩溃循环纳入看门狗重建；单槽异常与 tick 级 promise 拒绝互不影响。（#491）
 - Docker 槽创建可选 `machine.docker_cpus`（`0` 或省略不限；`0.01`–`16` 为 CPU 时间上限）：写入槽位记录并在容器创建/重建时生效（本地 `--cpus`、集群 `NanoCpus`），KVM 与非法值拒绝；不提供在线 PATCH 或把 `vcpus` 隐式映射为 Docker 上限；损坏落盘值在 reload/恢复/重建前拒绝。（#490）
 - 修复通知监控在冷却/无事件轮询时误清 `last_error`：仅在实际投递成功时清除，保留上次投递失败；解析 `dispatchNotify` 的 `ok: false` 与部分通道失败，同批次任一失败优先于成功事件。（#483）
 - 代理出口地理查询经 SOCKS5 时由代理端解析目标域名（查询路径内将 `socks5://` 规范为 `socks5h:`），避免依赖控制面本地 DNS；IPv4 与 IPv6 两阶段探测一致，不修改已保存代理 URI，HTTP(S) 代理与本地直连行为不变。（#482）

src/lib/cluster/remote-slot.mjs

@@ -667,6 +667,9 @@ export async function startRemoteSlot(vm, projectRoot, { recreate = false, routi
       existing.Config?.Labels?.['kin.vm.layout'] === SLOT_LAYOUT &&
       runtimeLabelOf(existing) === runtimeKind(vm) &&
       codexRoleMatches(existing, vm) &&
+      // Running stays true during Docker restart-policy backoff. Recreate the
+      // loop just like the local runtime; otherwise watchdog revival is a no-op.
+      !existing.State?.Restarting &&
       !recreate
     const common = {
       nodeId,

src/lib/transport/kernel-watchdog.mjs

@@ -19,8 +19,48 @@ import {
   readExistingKernelConfig,
   dataplaneUsesCcNode,
   WRAP_SLOT_MAX,
+  slotContainerName,
 } from './rust-kernel-supervisor.mjs'
 import { resolveInferenceEngine } from '../vm/slot-engine.mjs'
+import { slotHost } from '../vm/slot-host.mjs'
+import { dockerRequest } from '../cluster/docker-remote.mjs'
+
+/** A failed observation is not evidence of a dead container. Use the slot's own host, including KVM runners. */
+export async function observeWatchdogContainer(
+  vm,
+  { hostFor = slotHost, request = dockerRequest, timeoutMs = 1500 } = {},
+) {
+  let timer
+  try {
+    const name = slotContainerName({ vm })
+    if (!name) return { state: 'unknown' }
+    const observation = Promise.resolve().then(async () => {
+      const res = await request(hostFor(vm).dockerApi(), {
+        path: `/containers/${encodeURIComponent(name)}/json`,
+        timeoutMs,
+      })
+      if (res.status === 404) return { state: 'missing' }
+      if (res.status !== 200) return { state: 'unknown' }
+      const state = JSON.parse(res.body.toString('utf8'))?.State
+      if (state?.Paused === true) return { state: 'paused' }
+      if (state?.Restarting === true) return { state: 'restarting' }
+      if (state?.Running === true) return { state: 'running' }
+      if (state?.Running === false && ['created', 'exited', 'dead'].includes(state.Status)) return { state: 'stopped' }
+      return { state: 'unknown' }
+    })
+    // Covers connector/SSH setup too, before dockerRequest's own HTTP timeout starts.
+    return await Promise.race([
+      observation,
+      new Promise((resolve) => {
+        timer = setTimeout(() => resolve({ state: 'unknown' }), timeoutMs)
+      }),
+    ])
+  } catch {
+    return { state: 'unknown' }
+  } finally {
+    clearTimeout(timer)
+  }
+}
 
 export const DEFAULT_KERNEL_WATCHDOG = Object.freeze({
   enabled: true,
@@ -82,6 +122,7 @@ export function createKernelWatchdog({
   homeDirFor,
   ensure = ensureRustKernel,
   health = rustKernelHealth,
+  observeRuntime = observeWatchdogContainer,
   onFault = null,
   ensureTelemetry = null,
   telemetryEveryMs = TELEMETRY_HEAL_MS,
@@ -171,16 +212,33 @@ export function createKernelWatchdog({
       // Plain recovery keeps past attempts so a crash loop stays bounded by the window.
       if (kernelFaults.delete(vm.id)) restarts.delete(vm.id)
       // Config convergence, not a failure: outside the restart budget.
-      if (kernelSlotMismatch(exec)) await ensure(exec, { timeoutMs: config.timeout_ms })
+      if (kernelSlotMismatch(exec)) {
+        const observed = await observeRuntime(vm)
+        if (!held.has(vm.id) && observed?.state === 'running') await ensure(exec, { timeoutMs: config.timeout_ms })
+      }
       return
     }
     const reason = current?.unhealthy_reason || current?.error || 'kernel process down'
+    const observed = await observeRuntime(vm)
+    if (held.has(vm.id)) return
+    // Permission/transport failures and deliberately paused containers never authorize recovery.
+    if (!['running', 'missing', 'stopped', 'restarting'].includes(observed?.state)) return
+    if (observed.state !== 'running') {
+      if (vm.status !== 'running' || typeof revive !== 'function') return
+      if (!admitRestart(vm, `container ${observed.state}`, now())) return
+      const revived = await revive(vm)
+      if (!revived?.ok) console.warn(`[kernel-watchdog] ${vm.id} boot failed: ${revived?.error || revived?.code}`)
+      return
+    }
     if (!admitRestart(vm, reason, now())) return
     console.warn(`[kernel-watchdog] ${vm.id} restarting container: ${reason}`)
     const result = await ensure(exec, { timeoutMs: config.timeout_ms })
     if (held.has(vm.id)) return
     // The container itself is gone or exited: a kernel restart cannot help, a full boot can.
     if (result?.reason === 'container_unavailable' && typeof revive === 'function' && vm.status === 'running') {
+      // The container may disappear during ensure; do not mistake a daemon/permission error for absence.
+      const latest = await observeRuntime(vm)
+      if (held.has(vm.id) || !['missing', 'stopped', 'restarting'].includes(latest?.state)) return
       console.warn(`[kernel-watchdog] ${vm.id} container unavailable; booting slot`)
       const revived = await revive(vm).catch((error) => ({ ok: false, error: error?.message || String(error) }))
       if (!revived?.ok) console.warn(`[kernel-watchdog] ${vm.id} boot failed: ${revived?.error || revived?.code}`)
@@ -208,10 +266,15 @@ export function createKernelWatchdog({
         inflight.set(vm.id, work)
         try {
           await work
+        } catch {
+          // One failed probe/recovery must not starve later slots or reject an interval's promise.
+          console.warn(`[kernel-watchdog] ${vm.id} check failed; skipping this tick`)
         } finally {
           if (inflight.get(vm.id) === work) inflight.delete(vm.id)
         }
       }
+    } catch {
+      console.warn('[kernel-watchdog] tick failed; retrying at the next interval')
     } finally {
       inTick = false
     }

test/unit/kernel-watchdog-observation.test.mjs

@@ -0,0 +1,230 @@
+import assert from 'node:assert/strict'
+import test from 'node:test'
+import * as watchdog from '../../src/lib/transport/kernel-watchdog.mjs'
+import { kernelFaults } from '../../src/lib/transport/rust-kernel-client.mjs'
+
+test.afterEach(() => {
+  for (const id of ['vm-observation', 'vm-bad', 'vm-good']) kernelFaults.delete(id)
+})
+
+function fixture({ state = 'running', runtime, node_id, status = 'running', observe, health, ensure, revive } = {}) {
+  let at = 0
+  const actions = []
+  const vm = { id: 'vm-observation', status, runtime, node_id }
+  const wd = watchdog.createKernelWatchdog({
+    listTargets: () => [vm],
+    now: () => at,
+    config: { restart_backoff_sec: [10] },
+    health: health || (async () => null),
+    observeRuntime: observe || (async () => ({ state })),
+    ensure:
+      ensure ||
+      (async () => {
+        actions.push('ensure')
+        return { ok: true }
+      }),
+    revive:
+      revive ||
+      (async () => {
+        actions.push('revive')
+        return { ok: true }
+      }),
+  })
+  return {
+    wd,
+    vm,
+    actions,
+    advance: (ms) => {
+      at = ms
+    },
+  }
+}
+
+for (const state of ['unknown', 'paused']) {
+  test(`${state} runtime does not spend recovery budget or cause a restart`, async () => {
+    const f = fixture({ state })
+    for (const ms of [0, 10_000, 20_000]) {
+      f.advance(ms)
+      await f.wd.tick()
+    }
+    assert.deepEqual(f.actions, [])
+    assert.equal(f.wd.restartsInWindow(f.vm.id), 0)
+    assert.equal(kernelFaults.has(f.vm.id), false)
+  })
+}
+for (const state of ['missing', 'stopped', 'restarting']) {
+  for (const placement of [
+    { runtime: { type: 'docker' } },
+    { runtime: { type: 'kvm' } },
+    { node_id: 'node-fixture' },
+  ]) {
+    test(`${state} ${JSON.stringify(placement)} revives only after backoff, without a blind kernel restart`, async () => {
+      const f = fixture({ state, ...placement })
+      try {
+        await f.wd.tick()
+        assert.deepEqual(f.actions, [])
+        f.advance(10_000)
+        await f.wd.tick()
+        assert.deepEqual(f.actions, ['revive'])
+        assert.equal(f.wd.restartsInWindow(f.vm.id), 1)
+        f.advance(30_000)
+        await f.wd.tick()
+        assert.deepEqual(f.actions, ['revive'])
+        assert.ok(kernelFaults.has(f.vm.id))
+      } finally {
+        kernelFaults.delete(f.vm.id)
+      }
+    })
+  }
+}
+
+test('unknown observations recover without consuming retries; running-but-dead kernel is still restarted', async () => {
+  let state = 'unknown'
+  const f = fixture({ observe: async () => ({ state }) })
+  await f.wd.tick()
+  f.advance(30_000)
+  await f.wd.tick()
+  state = 'running'
+  await f.wd.tick()
+  assert.deepEqual(f.actions, [])
+  f.advance(40_000)
+  await f.wd.tick()
+  assert.deepEqual(f.actions, ['ensure'])
+})
+
+test('stopped intent is not probed or revived and planned intent is not booted', async () => {
+  for (const status of ['stopped', 'disabled', 'planned']) {
+    const f = fixture({ state: 'missing', status })
+    await f.wd.tick()
+    f.advance(10_000)
+    await f.wd.tick()
+    assert.deepEqual(f.actions, [])
+    assert.equal(f.wd.restartsInWindow(f.vm.id), 0)
+  }
+})
+
+test('a permission-like container_unavailable result cannot trigger boot without confirmed absence', async () => {
+  const f = fixture({
+    ensure: async () => ({ ok: false, reason: 'container_unavailable', error: 'daemon not running' }),
+  })
+  await f.wd.tick()
+  f.advance(10_000)
+  await f.wd.tick()
+  assert.deepEqual(f.actions, [])
+})
+
+test('hold during an observation prevents restart and releases the in-flight tick', async () => {
+  let finish
+  const f = fixture({
+    observe: () =>
+      new Promise((resolve) => {
+        finish = resolve
+      }),
+  })
+  const tick = f.wd.tick()
+  await new Promise((r) => setImmediate(r))
+  const held = f.wd.hold(f.vm.id)
+  finish({ state: 'missing' })
+  await Promise.all([tick, held])
+  f.advance(10_000)
+  await f.wd.tick()
+  assert.deepEqual(f.actions, [])
+})
+
+test('one rejected health probe or ensure call does not starve the next slot or reject tick', async () => {
+  for (const phase of ['health', 'ensure', 'observe']) {
+    let at = 0
+    const ensured = []
+    const wd = watchdog.createKernelWatchdog({
+      config: { restart_backoff_sec: [10] },
+      listTargets: () => [{ id: 'vm-bad' }, { id: 'vm-good' }],
+      now: () => at,
+      health: async (exec) => {
+        if (phase === 'health' && exec.vmId === 'vm-bad') throw new Error('fixture health')
+        return null
+      },
+      observeRuntime: async (vm) => {
+        if (phase === 'observe' && vm.id === 'vm-bad') throw new Error('fixture permission')
+        return { state: 'running' }
+      },
+      ensure: async (exec) => {
+        if (phase === 'ensure' && exec.vmId === 'vm-bad') throw new Error('fixture ensure')
+        ensured.push(exec.vmId)
+        return { ok: true }
+      },
+    })
+    await wd.tick()
+    at = 10_000
+    await wd.tick()
+    assert.ok(ensured.includes('vm-good'))
+  }
+})
+
+test('overlapping ticks do not duplicate observations or recovery', async () => {
+  let finish,
+    calls = 0
+  const f = fixture({
+    observe: () => {
+      calls++
+      return new Promise((r) => {
+        finish = r
+      })
+    },
+  })
+  const first = f.wd.tick()
+  await new Promise((r) => setImmediate(r))
+  await f.wd.tick()
+  assert.equal(calls, 1)
+  finish({ state: 'running' })
+  await first
+})
+
+test('Docker observation distinguishes 404, HTTP failures, malformed payloads and state flags on the selected host', async () => {
+  const cases = [
+    [404, {}, 'missing'],
+    [403, {}, 'unknown'],
+    [500, {}, 'unknown'],
+    [200, {}, 'unknown'],
+    [200, { State: { Running: true } }, 'running'],
+    [200, { State: { Running: true, Paused: true } }, 'paused'],
+    [200, { State: { Running: true, Restarting: true } }, 'restarting'],
+    [200, { State: { Running: false, Status: 'exited' } }, 'stopped'],
+    [200, { State: { Running: false } }, 'unknown'],
+  ]
+  for (const [status, payload, state] of cases) {
+    const vm = { id: 'vm-fixture', node_id: 'node-fixture', runtime: { container: 'custom/name' } }
+    const connector = () => {
+      throw new Error('fixture must not connect')
+    }
+    const result = await watchdog.observeWatchdogContainer(vm, {
+      hostFor: (target) => {
+        assert.equal(target, vm)
+        return { dockerApi: () => connector }
+      },
+      request: async (connect, options) => {
+        assert.equal(connect, connector)
+        assert.equal(options.path, '/containers/custom%2Fname/json')
+        return { status, body: Buffer.from(JSON.stringify(payload)) }
+      },
+    })
+    assert.equal(result.state, state)
+  }
+  const options = { hostFor: () => ({ dockerApi: () => () => {} }) }
+  for (const request of [
+    async () => {
+      throw new Error('EACCES')
+    },
+    async () => ({ status: 200, body: Buffer.from('broken json') }),
+  ]) {
+    assert.equal((await watchdog.observeWatchdogContainer({ id: 'vm-01' }, { ...options, request })).state, 'unknown')
+  }
+  assert.equal(
+    (
+      await watchdog.observeWatchdogContainer(
+        { id: 'vm-01' },
+        { ...options, request: () => new Promise(() => {}), timeoutMs: 10 },
+      )
+    ).state,
+    'unknown',
+  )
+})