dsh-task-watchdog

August 25, 2026 · View on GitHub

DSH（DeepSeek Harness）通用任务健康监控插件：DSH 插件把自己的长轮询/后台任务登记到 ctx.watchdog，由它统一做心跳检测、根因诊断、自动重启、失败告警、过程审计——杜绝"进程活着但任务静默停摆"。

为什么需要它：DSH 插件的长轮询任务（IM 通道、MCP、定时任务等）常见故障是 poll 循环抛异常被吞、start 卡死、依赖服务晚加载——任务不再收消息但没有任何告警。dsh-relay 曾因此漏读用户回复（iMessage 水位卡死）。只做自动重启是掩盖问题：本插件每次停滞都先抓根因诊断快照，再恢复，并把整个过程记录成事件流。

设计：对接 DSH 原生 ctx.jobs（不重复造轮子）

任务登记/生命周期/取消/完成通知全部复用 DSH 宿主自带的 ctx.jobs：

消费方插件                          dsh-task-watchdog (宿主级)
─────────────────────              ─────────────────────────
ctx.jobs.start() ──jobId──►        ctx.jobs.get() 判生命周期
循环里 wd.beat(jobId) ──心跳──►    停滞检测（stallSecs）
                                  ↓ 诊断快照（pid/心跳/日志）→ persist
wd.monitor({jobId, restart,       ↓ 调 restart(jobId) ← 消费方注入
  persist, alertSink})            ↓ 失败计数 → 告警 alertSink
消费方 jobs.kill() ──►             任务终结自动移除监控（=主动停止）

watchdog 不认任何具体任务类型（imessage/wechat/email/MCP/定时任务…），只认 jobId + 消费方注入的恢复回调——完全通用。

能力

功能	说明
登记监控	`watchdog.monitor({jobId, label?, spawn?, check?, persist?, alertSink?})`——监控一个已用 `ctx.jobs.start` 创建的任务
心跳上报	任务循环里 `watchdog.beat(jobId)`；或提供 `check()` 自定义健康探测
停滞检测	心跳超过 `stallSecs` 未更新，或 `check()` 返回不健康 → 判定卡死
根因诊断	停滞时抓诊断快照（进程状态/各任务心跳/最近日志），经 `persist` 持久化
通用恢复	消费方提供 `spawn()`（怎么建一个新任务）；watchdog 全权处理 `jobs.kill(旧) → spawn() → 自动重新 monitor 新 jobId`
失败告警	连续失败超 `maxRestarts` 次 → 经其他活任务的 `alertSink` 推送
主动停止区分	任务终结（消费方 `jobs.kill`）→ 自动移除监控；`spawn` 返回空（消费方决定不重建）→ 停止监控；意外停滞（心跳丢失且仍 running）→ 诊断+恢复
事件流	`watchdog.onEvent(listener)` 订阅 / `watchdog.events()` 查历史——发现问题→追踪→解决全过程可观测
状态查询	`watchdog.status()` / `watchdog.diagnostics()` 供调试

事件流（审计 / 通知接口）

每次动作发出结构化事件 { seq, ts, type, jobId, label, reason?, ... }：

type	含义
`registered`	任务开始被监控
`diagnosed`	发现问题（含完整诊断快照 snapshot）
`restarting`	开始恢复（第 N 次尝试）
`restarted`	恢复完成
`recovered`	恢复成功（心跳回归）
`alerted`	告警（连续失败 / restart 失败）
`unwatched`	移除监控（主动停止 / 任务终结）
`terminated`	任务终结（completed/killed/failed）

对接任意通讯方式（iMessage / 邮件 / webhook / 其他应用，用于审核与记录）：

const wd = ctx.get('watchdog')
const unsub = wd.onEvent((ev) => {
  // 例 1：转发到你的通知通道（webhook / iMessage / 邮件）
  myNotify.send(`[watchdog] ${ev.ts} ${ev.type} ${ev.jobId} ${ev.reason ?? ''}`)
  // 例 2：写入审计日志
  myAudit.append(ev)
})
// 不再需要时：unsub()

历史查询（复盘/审核）：watchdog.events(limit?) 返回最近事件（默认 100，最多 500）。

安装

# 1. 在 profile 的 package.json dependencies 加入：
#    "dsh-task-watchdog": "github:<owner>/dsh-task-watchdog"
#    或本地： "dsh-task-watchdog": "link:/Users/<you>/.dsh/plugins/dsh-task-watchdog"
# 2. 加入 dsh.profile.bundles 列表
# 3. 在 profile 的 cordis.patch.yml 插入：
#    - id: dsh-task-watchdog
#      name: 'dsh-task-watchdog'
#      config:
#        enabled: true
#        watchdogSecs: 30      # 巡检周期（秒），0=关闭
#        stallSecs: 90         # 心跳停滞阈值（秒）
#        maxRestarts: 3        # 连续失败多少次后停止重试并告警
#        diagKeep: 10          # 诊断快照保留条数
# 4. pnpm install && 重启 DSH

消费方用法（供其他插件）

// 消费方插件的 apply：
export const inject = ['jobs']   // 你仍需用 ctx.jobs 建任务

function myApply(ctx) {
  const wd = ctx.get('watchdog')
  // 1. 定义"怎么建一个任务"（spawn：恢复机制用，返回新 jobId）
  const spawn = () => {
    const jid = ctx.jobs.start({
      kind: 'my-plugin',
      label: '我的轮询任务',
      run: () => { /* ... 返回 hooks */ },
    })
    return jid
  }
  // 2. 建首个任务并登记监控。恢复由 watchdog 全权处理：
  //    停滞 → jobs.kill(旧) → spawn() 重建 → 自动重新 monitor 新 jobId。
  //    消费方不需要写任何恢复逻辑。
  const unmon = wd?.monitor({
    jobId: spawn(),
    spawn,                 // watchdog 恢复时调用，返回新 jobId
    persist: (diag) => { /* 诊断落盘到你的 state */ },
    alertSink: (msg) => { /* 通过你的通道推送告警 */ },
  })
  ctx.effect?.(() => unmon?.())

  // 3. 任务循环里上报心跳
  while (running) {
    wd?.beat(jobId)
    // ... 实际轮询 ...
  }
}

测试

npm test    # 单元测试 + apply 冒烟（对接 ctx.jobs 契约）

覆盖：monitor/beat/stop 基本行为、心跳停滞触发诊断+恢复+失败计数、spawn 通用恢复机制（kill→重建→重新监控、返回空=停止监控）、任务终结自动移除监控（主动停止 vs 意外停滞）、check() 不健康路径、恢复健康清失败计数、事件流（订阅/历史/类型）、参数校验。

许可

MIT

测试与覆盖

测试：npm test（统一运行器 test/run-all.mjs：dry-run / env-check / 契约）
覆盖矩阵与质量检测基线：test/DRYRUN.md（正向推演：功能规格 → 测试设计技术 → 规模层 → 风险 → 断言 → 回归实证；含 c8 覆盖率快照与盲区标注）
覆盖率复测：npm run coverage（c8 text-summary；数字变化时请同步更新 DRYRUN.md）