fix(proc): 子进程崩溃自愈 + 集中台账 + 注册面摘除

根因:子进程插件被 kill 后,内核只发了一个无人订阅的事件,
工具/stage handler/IO 通道全留在注册表里指向死进程,
模型继续调用只吃 ErrProcessExited,没有任何路径把插件拉回来。

## 四层修复

### 1. 专职 waitLoop(进程收割)
- 每个子进程配一根 waitLoop goroutine,是 cmd.Wait() 的唯一调用点
- 不再依赖 stdout EOF 判定死亡(孙子进程继承 stdout 时 EOF 永不到来)
- 手工 os.Pipe 替代 cmd.StdinPipe/StdoutPipe,避免 waitLoop 与
  os/exec 的内部关闭竞争
- host.go: Host.Supervisor(),Host.Close() 先 StopAll 再拆段

### 2. 集中台账 Supervisor
- proc/supervisor.go: 插件 Spawn 握手成功即 track,进程退出即 untrack
- StopAll: 并发发 plugin.stop 走优雅路径,到期仍在的一律 Kill
- 关停后才完成握手的进程被立即结束,不会活过内核
- 消除「孤儿进程持共享段映射 → SIGBUS」的隐患

### 3. 注册面摘除(detachPlugin)
- 新增 StageHost.UnregisterPluginStages:摘除指定插件的全部 stage handler
- 新增 Registry.pluginChannels 台账:记录每个插件注册的 IO 通道
- 三条路径统一走 detachPlugin:Disable / ReloadOne / RemovePlugin
- StopAndUnload 漏了 IO 通道也一并补上

### 4. 自动重启
- onProcCrash 从「只发事件」改为「摘注册面 → 从注册表移除 → 异步排重启」
- scheduleProcRestart: 窗口 5 分钟内最多 3 次,线性退避 1s/2s/3s
- 超限停手留日志;重启前复核是否已被 Disable 或被其他路径加载
- 崩溃计数窗口过期自动归零

### 5. 主动停止 vs 崩溃的区分
- proc.Plugin 新增 stopping 标志:Stop()/Close() 里 Set(true)
- handleExit 读 stopping 标志,主动停止不上报 onCrash
- 防止重载/禁用/卸载被误判为崩溃触发多余重启

### 6. Linux Pdeathsig 兜底
- procattr_linux.go: SysProcAttr.Pdeathsig = SIGKILL
- 兜 homed 自身被 SIGKILL/OOM 时子进程变孤儿的场景
- macOS/Windows 无等价物,空实现

### 7. pluginmgr 升级
- PluginManager 接口新增 PluginRuntime / ListPluginRuntimes
- plugin_list 输出运行态:loaded / alive / pid / crash_count / channel
- 新增 plugin_status: 全量运行期快照 + dead/unhealthy 汇总
- 新增 plugin_restart: 无条件重启单个插件(plgreload 不动未改二进制的插件)

### 测试
- process_test.go: 3 例(grandchild stdout 感知 / Supervisor track-untrack /
  StopAll 无孤儿)
- crash_recovery_test.go: 8 例(detach 三项齐全 / 通道重注册 / 崩溃不阻塞 /
  退避阈值 / 窗口过期 / 关停中跳过 / PluginRuntime 通道识别)
- stages_plugin_test.go: 4 例(stage 按插件摘除 / 空 stage 清理 / 空名 no-op /
  工具+stage 双摘后可重新注册同名)
This commit is contained in:
JianFeeeee
2026-09-03 12:37:58 +08:00
parent 351e99dd70
commit 02cc74ce11
18 changed files with 1649 additions and 74 deletions

View File

@ -14,14 +14,25 @@ type StageHost struct {
toolDefs []sdk.ToolDef
tools map[string]sdk.ToolHandler
toolPlugins map[string]string
stages map[sdk.Stage][]sdk.StageHandler
stages map[sdk.Stage][]stageEntry
}
// stageEntry 把 stage handler 与它的归属插件绑定。
//
// 为何需要归属:子进程插件崩溃后,它注册的 handler 闭包仍在这张表里,
// 每次 RunStage 都会经 RPC 打向已死进程并报 ErrProcessExited重启后新 handler
// 又追加进来,旧的永不退场——错误与重复执行随重启次数线性累积。
// 有了归属才能在卸载/崩溃时成组摘除。
type stageEntry struct {
plugin string
fn sdk.StageHandler
}
func NewStageHost() *StageHost {
return &StageHost{
tools: make(map[string]sdk.ToolHandler),
toolPlugins: make(map[string]string),
stages: make(map[sdk.Stage][]sdk.StageHandler),
stages: make(map[sdk.Stage][]stageEntry),
}
}
@ -41,9 +52,15 @@ func (h *StageHost) RegisterTool(name string, def sdk.ToolDef, handler sdk.ToolH
}
func (h *StageHost) RegisterStage(stage sdk.Stage, handler sdk.StageHandler) {
h.RegisterStageFor("", stage, handler)
}
// RegisterStageFor 注册带归属插件名的 stage handler。
// plugin 为空时等同 RegisterStage内核自身注册的 handler不参与成组摘除
func (h *StageHost) RegisterStageFor(plugin string, stage sdk.Stage, handler sdk.StageHandler) {
h.mu.Lock()
defer h.mu.Unlock()
h.stages[stage] = append(h.stages[stage], handler)
h.stages[stage] = append(h.stages[stage], stageEntry{plugin: plugin, fn: handler})
}
func (h *StageHost) GetToolDefs() []sdk.ToolDef {
@ -106,6 +123,36 @@ func (h *StageHost) UnregisterPluginTools(pluginName string) {
h.toolDefs = keepDefs
}
// UnregisterPluginStages 摘除某插件注册的全部 stage handler返回摘除数量。
//
// 与 UnregisterPluginTools 成对:卸载/重载/崩溃时两者都得做,
// 否则插件的工具没了但 stage handler 还在,继续打向不存在的插件。
func (h *StageHost) UnregisterPluginStages(pluginName string) int {
if pluginName == "" {
return 0
}
h.mu.Lock()
defer h.mu.Unlock()
removed := 0
for stage, entries := range h.stages {
keep := entries[:0:0]
for _, e := range entries {
if e.plugin == pluginName {
removed++
continue
}
keep = append(keep, e)
}
if len(keep) == 0 {
delete(h.stages, stage)
continue
}
h.stages[stage] = keep
}
return removed
}
func inferToolPlugin(name string) string {
for i := 0; i < len(name); i++ {
if name[i] == '_' {
@ -123,14 +170,15 @@ func inferToolPlugin(name string) string {
// handler 返回的 error 会被收集到 ctx.Errors 中并记录日志,不会中断其他 handler 的执行。
func (h *StageHost) RunStage(stage sdk.Stage, ctx *sdk.StageContext) {
h.mu.RLock()
handlers := h.stages[stage]
entries := make([]stageEntry, len(h.stages[stage]))
copy(entries, h.stages[stage])
h.mu.RUnlock()
if len(handlers) == 0 {
if len(entries) == 0 {
return
}
var wg sync.WaitGroup
errCh := make(chan error, len(handlers))
for _, handler := range handlers {
errCh := make(chan error, len(entries))
for _, entry := range entries {
wg.Add(1)
go func(fn sdk.StageHandler) {
defer wg.Done()
@ -142,7 +190,7 @@ func (h *StageHost) RunStage(stage sdk.Stage, ctx *sdk.StageContext) {
if err := fn(ctx); err != nil {
errCh <- err
}
}(handler)
}(entry.fn)
}
wg.Wait()
close(errCh)

View File

@ -0,0 +1,96 @@
package core
import (
"testing"
sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk"
)
// stage handler 必须能按插件成组摘除。
//
// 修复前 StageHost.stages 只存匿名函数,没有归属信息:
// 插件崩溃/卸载后它的 handler 永远留在表里,每轮 RunStage 都被并发调起并
// 打向已死进程;重启后新 handler 追加进来,旧的仍不退场——
// 错误与重复执行随重启次数线性累积。
func TestUnregisterPluginStages_RemovesOnlyThatPlugin(t *testing.T) {
h := NewStageHost()
var aRan, bRan, coreRan int
h.RegisterStageFor("a", sdk.StagePreAction, func(*sdk.StageContext) error { aRan++; return nil })
h.RegisterStageFor("b", sdk.StagePreAction, func(*sdk.StageContext) error { bRan++; return nil })
// 内核自身注册的 handler无归属不该被插件摘除波及
h.RegisterStage(sdk.StagePreAction, func(*sdk.StageContext) error { coreRan++; return nil })
h.RunStage(sdk.StagePreAction, &sdk.StageContext{})
if aRan != 1 || bRan != 1 || coreRan != 1 {
t.Fatalf("首轮应全部执行a=%d b=%d core=%d", aRan, bRan, coreRan)
}
if n := h.UnregisterPluginStages("a"); n != 1 {
t.Errorf("应摘除 1 个 handler实际 %d", n)
}
h.RunStage(sdk.StagePreAction, &sdk.StageContext{})
if aRan != 1 {
t.Errorf("已摘除的插件 handler 不该再被调用,实际执行 %d 次", aRan)
}
if bRan != 2 || coreRan != 2 {
t.Errorf("其他 handler 应照常执行b=%d core=%d", bRan, coreRan)
}
}
// 摘除某插件的最后一个 handler 后,该 stage 应从表中消失RunStage 直接短路)。
func TestUnregisterPluginStages_DropsEmptyStage(t *testing.T) {
h := NewStageHost()
h.RegisterStageFor("solo", sdk.StageAfterToolcall, func(*sdk.StageContext) error { return nil })
if n := h.UnregisterPluginStages("solo"); n != 1 {
t.Fatalf("应摘除 1 个,实际 %d", n)
}
h.mu.RLock()
_, exists := h.stages[sdk.StageAfterToolcall]
h.mu.RUnlock()
if exists {
t.Error("stage 已无 handler 时应从表中删除")
}
}
// 空插件名不得误摘内核自身注册的 handler。
func TestUnregisterPluginStages_EmptyNameIsNoop(t *testing.T) {
h := NewStageHost()
ran := 0
h.RegisterStage(sdk.StagePreAction, func(*sdk.StageContext) error { ran++; return nil })
if n := h.UnregisterPluginStages(""); n != 0 {
t.Errorf("空插件名应是 no-op实际摘除 %d", n)
}
h.RunStage(sdk.StagePreAction, &sdk.StageContext{})
if ran != 1 {
t.Errorf("内核 handler 应保留并执行,实际 %d 次", ran)
}
}
// 工具与 stage 的摘除互不干扰:都摘完后两者皆空。
func TestUnregisterPluginToolsAndStages_Together(t *testing.T) {
h := NewStageHost()
if err := h.RegisterTool("demo_run", sdk.ToolDef{Name: "demo_run", Plugin: "demo"},
func(map[string]interface{}) (interface{}, error) { return nil, nil }); err != nil {
t.Fatal(err)
}
h.RegisterStageFor("demo", sdk.StagePreAction, func(*sdk.StageContext) error { return nil })
h.UnregisterPluginTools("demo")
h.UnregisterPluginStages("demo")
if h.ToolCount() != 0 {
t.Errorf("工具应已摘除,实际 %d", h.ToolCount())
}
if h.ToolPlugin("demo_run") != "" {
t.Error("工具→插件映射应清空")
}
// 摘除后可重新注册同名工具(重启路径的前提)
if err := h.RegisterTool("demo_run", sdk.ToolDef{Name: "demo_run", Plugin: "demo"},
func(map[string]interface{}) (interface{}, error) { return nil, nil }); err != nil {
t.Errorf("摘除后应可重新注册同名工具,实际: %v", err)
}
}