From 9de3b365a6404ab64f024bcf843a5f5de74ac33a Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Thu, 17 Sep 2026 20:24:03 +0800 Subject: [PATCH] =?UTF-8?q?fix(agentcli):=20=E4=BF=AE=204=20=E4=B8=AA?= =?UTF-8?q?=E7=9C=9F=E5=AE=9E=E7=BC=BA=E9=99=B7=E2=80=94=E2=80=94=E5=81=9C?= =?UTF-8?q?=E6=9C=BA=E6=AD=BB=E9=94=81=E3=80=81=E8=B6=85=E6=97=B6=E6=B3=84?= =?UTF-8?q?=E6=BC=8F=E3=80=81=E5=83=B5=E5=B0=B8=E5=A0=86=E7=A7=AF=E3=80=81?= =?UTF-8?q?=E5=AD=99=E8=BF=9B=E7=A8=8B=E9=80=83=E9=80=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit jianf 提示 agentcli 可能有问题，系统性审了一遍（含 -race 与线上实证）， 确认并修复 4 个互相叠加的真实缺陷，每个都配了「去掉修复即失败」的回归测试。 1) 停机/热重载死锁（plugin_stop_test.go） Stop() 先 p.wg.Wait() 再 Close 终端，而 readLoop 自己也记在 p.wg 上、 只监 t.stopCh 不监 p.stopCh。只要有一个终端开着，wg.Wait() 就永不返回。 后果：插件卸载/热重载（StopAndUnload/ReloadOne）与停机全挂死，且 registry 持锁时是整个内核一起挂。 修：先关活跃终端（move 出 map 后在锁外 Close），再 wg.Wait()； readLoop 顶部加 p.stopCh 探测；Stop() 用 sync.Once 保证幂等。 2) 终端超时后资源全泄漏（plugin_lifecycle_test.go） readLoop 的 IsExpired 分支只 delete(sessions) 后 return，既不 Kill 也不 Close。终端已被移出 sessions，cleanupLoop 也再看不到它，进程/PTY fd/ reader 协程无人回收。实测：timeout=1s 的 sleep 300 超时后进程仍在跑。 修：readLoop 加 defer releaseResources()，保证「只要退出就释放」。 3) 子进程从不回收 → 僵尸堆积（pty_linux.go + plugin_lifecycle_test.go） newCommandPty 只 Start 从不 Wait。线上实测 homed 名下已有一个 [sh] 僵尸子进程。 修：linuxPty 加 Wait()（sync.Once 保证只 Wait 一次）， releaseResources 通过可选接口 Wait() error 调用（Windows ConPTY 不实现则跳过）。 4) Kill 只杀直接子进程，孙进程逃逸（pty_linux.go + plugin_lifecycle_test.go） newCommandPty 用 Setsid，sh 是新进程组领头，真正的命令（sleep/vim）是 其孙进程且同组。只 Kill(sh) 会留下孤儿继续跑。实测：`sleep 300; echo done` 只杀 leader 后 sleep 仍在（被 init 收养）。 修：改为 syscall.Kill(-pid, SIGKILL) 杀整个进程组，失败再回落单进程 Kill。 测试设计要点：回归用例必须让「sh 保留为父进程 + 孙进程显式 trap "" HUP」， 否则单个 sleep 会被 sh exec 掉、关 PTY 的 SIGHUP 又会顺手带走孙进程， 两个缺陷都测不出来（这两种情况都实际踩过并修正了用例）。 全量 go test ./internal/... ./cmd/... 通过，agentcli 单包 -race 通过。 --- internal/plugins/agentcli/plugin.go | 88 +++++++++++-- .../plugins/agentcli/plugin_lifecycle_test.go | 120 ++++++++++++++++++ internal/plugins/agentcli/plugin_stop_test.go | 71 +++++++++++ internal/plugins/agentcli/pty_linux.go | 30 ++++- 4 files changed, 297 insertions(+), 12 deletions(-) create mode 100644 internal/plugins/agentcli/plugin_lifecycle_test.go create mode 100644 internal/plugins/agentcli/plugin_stop_test.go diff --git a/internal/plugins/agentcli/plugin.go b/internal/plugins/agentcli/plugin.go index e80dcd0..59b2ffb 100644 --- a/internal/plugins/agentcli/plugin.go +++ b/internal/plugins/agentcli/plugin.go @@ -75,6 +75,11 @@ type TerminalSession struct { backoff time.Duration // 输出风暴退避：持续高速输出时通知间隔翻倍 watch terminalWatch // 该终端的提醒规则 + // resourcesReleased 标记 releaseResources 是否已执行（幂等保护）。 + // 不与 closed 合用：closed 语义是「用户主动要求关闭」，releaseResources + // 是「后端资源已释放」，readLoop 自然退出时只后者为真。 + resourcesReleased bool + // 实时画面推流（terminal_output 事件） stream bytes.Buffer // 待推送的增量输出，由 readLoop 每 200ms flush 一次 } @@ -101,12 +106,9 @@ func (t *TerminalSession) Close() { t.mu.Unlock() close(t.stopCh) - // 先终止进程（各平台实现：Linux 信号 / Windows TerminateProcess，幂等），再释放资源。 - // 不能依赖 cmd.Process.Kill()：Windows 后端 cmd.Process 为占位（仅 Pid）。 - if t.session != nil { - _ = t.session.Kill() - } - t.session.Close() + // 终止进程并释放 PTY（幂等；readLoop 自然退出时已调过就直接返回）。 + t.releaseResources() + // 等 readLoop 走完退出流程（它会发最后的 output/停止事件）。 <-t.done } @@ -158,6 +160,7 @@ type Plugin struct { mu sync.Mutex wg sync.WaitGroup stopCh chan struct{} + stopOnce sync.Once sessions map[string]*TerminalSession nextID int defaultTimeout time.Duration @@ -415,17 +418,38 @@ func (p *Plugin) Start(s *sdk.PluginSDK) error { } func (p *Plugin) Stop() error { - close(p.stopCh) - p.wg.Wait() + // 幂等：Stop 可能被多条路径调到（StopAndUnload 后再 StopAll、 + // 停机与热重载交错）。close 一个已关闭的 channel 会 panic， + // 故用 Once 兜住。 + p.stopOnce.Do(func() { + close(p.stopCh) + p.shutdown() + }) + return nil +} +func (p *Plugin) shutdown() { + // 必须**先**关掉活跃终端，再等 wg。 + // + // 为什么顺序不能反：readLoop 自己也记在 p.wg 上，而它的退出条件 + // 是「t.stopCh 收到信号」或「进程自行结束／超时」——它不监 p.stopCh。 + // 旧实现在这里先 p.wg.Wait() 才 Close 终端：只要还有任何一个终端开着， + // readLoop 永远等不到 t.stopCh，wg.Wait() 就永返回不了。 + // 后果是插件卸载 / 热重载（StopAndUnload / ReloadOne）与停机全挂在 + // 这一步，且持有 registry 锁时就是全内核一起挂。 p.mu.Lock() + sessions := make([]*TerminalSession, 0, len(p.sessions)) for _, t := range p.sessions { - t.Close() + sessions = append(sessions, t) } p.sessions = nil p.mu.Unlock() - return nil + for _, t := range sessions { + t.Close() + } + + p.wg.Wait() } func (p *Plugin) nextIDLocked() string { @@ -851,10 +875,46 @@ func emitTermState(s *sdk.PluginSDK, t *TerminalSession, running bool) { }) } +// releaseResources 幂等地释放终端后端：杀进程 + 关 PTY。 +// +// 为何需要单独一个方法：readLoop 是终端自然的退出点（超时/进程结束/ +// 读取错误/插件停机），但 close(t.done) 的时机意味着它**不能**调 +// TerminalSession.Close()——后者会 <-t.done 等 readLoop 退出，而 readLoop +// 正在自己里面，直接死锁。所以这里只做「不再需要 readLoop 配合」的那半： +// 终止进程与释放 fd。 +func (t *TerminalSession) releaseResources() { + t.mu.Lock() + if t.resourcesReleased { + t.mu.Unlock() + return + } + t.resourcesReleased = true + t.mu.Unlock() + + if t.session != nil { + _ = t.session.Kill() + // 回收子进程（避免僵尸）。后端可选实现：Linux PTY 在 Kill 后 + // 必须 Wait 才能把 清掉；不实现的后端（如 Windows + // ConPTY）跳过即可。 + if reaper, ok := t.session.(interface{ Wait() error }); ok { + _ = reaper.Wait() + } + _ = t.session.Close() + } +} + func (p *Plugin) readLoop(t *TerminalSession, s *sdk.PluginSDK) { defer p.wg.Done() defer close(t.done) + // 无论从哪个分支退出，都释放进程与 PTY。 + // + // 旧实现只在「超时」和「进程退出」两个分支 delete(sessions) 后直接 + // return：超时分支完全不碰 session，一个 sleep 999 超时后进程、PTY fd + // 与 reader 協程全数泄漏（readLoop 已经从 sessions 里删掉了，cleanupLoop + // 也再看不到它，没人能回收）。defer 保证「只要退出就释放」。 + defer t.releaseResources() + // reader 协程独享这个读缓冲：结果随 readResult 携带， // readLoop 不再从其中做 copy（见 reader 注释，那是对共享缓冲 // 的并发读写，-race 实测触发）。 @@ -896,6 +956,14 @@ func (p *Plugin) readLoop(t *TerminalSession, s *sdk.PluginSDK) { quietLatency := 2 * time.Second for { + // p.stopCh：插件停机。readLoop 记在 p.wg 上，若不在此退出， + // Stop() 的 wg.Wait() 就只能等终端自己超时（最长 30 分钟）。 + select { + case <-p.stopCh: + return + default: + } + if t.IsExpired() { log.Printf("[agentcli] terminal %s expired after %v", t.id, t.timeout) s.InjectTextOpts("agentcli", "agentcli", fmt.Sprintf("[终端 %s 已超时关闭（%s）]", t.id, t.timeout), diff --git a/internal/plugins/agentcli/plugin_lifecycle_test.go b/internal/plugins/agentcli/plugin_lifecycle_test.go new file mode 100644 index 0000000..c4ac1a0 --- /dev/null +++ b/internal/plugins/agentcli/plugin_lifecycle_test.go @@ -0,0 +1,120 @@ +//go:build linux + +package agentcli + +import ( + "os" + "strconv" + "strings" + "syscall" + "testing" + "time" + + sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk" +) + +// 超时终端必须真正释放：杀掉整个进程组（包括 sh 的子进程如 sleep） +// 且回收子进程（不留 ）。 +// +// 旧实现三个缺陷叠加： +// 1. readLoop 的 IsExpired 分支只 delete(sessions) 后 return，不 Kill 不 Close； +// 2. Kill 只杀直接子进程 sh，Setsid 后真正的命令（sleep）是孙进程，成为孤儿； +// 3. 只 Start 从不 Wait，退出的子进程无人回收，积成僵尸。 +func TestExpiredTerminalReleasesProcessGroup(t *testing.T) { + p := New("agentcli") + sdkInst := sdk.New("agentcli", sdk.SDKConfig{ + RegTool: func(string, sdk.ToolDef, sdk.ToolHandler) error { return nil }, + RegStage: func(sdk.Stage, sdk.StageHandler) {}, + RegAPI: func(string) error { return nil }, + Settings: sdk.NewSettings("agentcli", nil), + }) + sdkInst.SetIOInjector(&injectCapture{}) + if err := p.Start(sdkInst); err != nil { + t.Fatal(err) + } + defer p.Stop() + + // 命令设计要点（缺一不可）： + // 1) 让 sh 保留为父进程、另起孙进程（`... & wait`）——单个 `sleep 300` + // 会被 sh 直接 exec 掉，只有一个进程，测不到「孙进程逃逸」； + // 2) 孙进程显式忽略 SIGHUP——否则关 PTY master 时内核发的 SIGHUP 会 + // 顺手把它带走，于是「只杀 leader」也能通过，测不出进程组 Kill 的必要性。 + // 两个条件合起来，只有给整个进程组发 SIGKILL 才能清干净。 + marker := `(trap "" HUP; sleep 300) & wait` + res, err := p.handleCreate(sdkInst, map[string]interface{}{ + "command": marker, + "timeout": "1s", + }) + if err != nil { + t.Fatal(err) + } + m := res.(map[string]interface{}) + if m["error"] != nil { + t.Skipf("PTY unavailable: %v", m["error"]) + } + + p.mu.Lock() + ts := p.sessions[m["id"].(string)] + p.mu.Unlock() + if ts == nil { + t.Fatal("terminal not registered") + } + pid := ts.cmd.Process.Pid + + // 等超时被 readLoop 处理（含 Kill 进程组 + Wait 回收） + deadline := time.Now().Add(8 * time.Second) + for time.Now().Before(deadline) { + if !procAlive(pid) && !procGroupAlive(pid) { + return + } + time.Sleep(200 * time.Millisecond) + } + t.Fatalf("expired terminal not released: leader pid=%d alive=%v groupAlive=%v", + pid, procAlive(pid), procGroupAlive(pid)) +} + +func procAlive(pid int) bool { return syscall.Kill(pid, 0) == nil } + +// procGroupAlive 检查「原进程组」里是否还有活着的成员（含被 init 收养的孙进程）。 +// +// 不能用 kill(-pgid, 0)：组领头进程一死，内核就可能回收该 pgid， +// 即便组里还有被 reparent 的成员，这个探测也会失败。 +// 改为直接遍历 /proc 查 pgid 匹配的活进程。 +func procGroupAlive(pgid int) bool { + entries, err := os.ReadDir("/proc") + if err != nil { + return false + } + for _, e := range entries { + if pid, err := strconv.Atoi(e.Name()); err == nil { + if readPgid(pid) == pgid { + return true + } + } + } + return false +} + +// readPgid 从 /proc//stat 读进程组 id（第 5 个字段）。 +// stat 的 comm 字段可能含空格/括号，所以从最后一个 ')' 之后再切分。 +func readPgid(pid int) int { + data, err := os.ReadFile("/proc/" + strconv.Itoa(pid) + "/stat") + if err != nil { + return -1 + } + s := string(data) + i := strings.LastIndex(s, ")") + if i < 0 { + return -1 + } + fields := strings.Fields(s[i+1:]) + if len(fields) < 3 { + return -1 + } + // fields[0]=state, [1]=ppid, [2]=pgrp + pgid, err := strconv.Atoi(fields[2]) + if err != nil { + return -1 + } + return pgid +} diff --git a/internal/plugins/agentcli/plugin_stop_test.go b/internal/plugins/agentcli/plugin_stop_test.go new file mode 100644 index 0000000..06c1ca6 --- /dev/null +++ b/internal/plugins/agentcli/plugin_stop_test.go @@ -0,0 +1,71 @@ +//go:build linux || windows + +package agentcli + +import ( + "testing" + "time" + + sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk" +) + +// Stop() 必须在有活跃终端时也能返回：它先 p.wg.Wait() 再 Close 终端， +// 而 readLoop 自己也记在 p.wg 上且不监听 p.stopCh —— 若 readLoop 只在 +// t.stopCh 上阻塞，p.wg.Wait() 会永远等下去（死锁）。 +func TestStopWithActiveTerminal(t *testing.T) { + p := New("agentcli") + sdkInst := sdk.New("agentcli", sdk.SDKConfig{ + RegTool: func(string, sdk.ToolDef, sdk.ToolHandler) error { return nil }, + RegStage: func(sdk.Stage, sdk.StageHandler) {}, + RegAPI: func(string) error { return nil }, + Settings: sdk.NewSettings("agentcli", nil), + }) + sdkInst.SetIOInjector(&injectCapture{}) + if err := p.Start(sdkInst); err != nil { + t.Fatal(err) + } + + term := newMockTerm() + ts := newTestSession(term) + ts.command = "sleep 999" + p.mu.Lock() + p.sessions[ts.id] = ts + p.mu.Unlock() + startReadLoop(p, sdkInst, ts) + + time.Sleep(100 * time.Millisecond) + + done := make(chan struct{}) + go func() { p.Stop(); close(done) }() + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("Stop() deadlocked with an active terminal") + } +} + +// Stop 必须幂等：重复调用不能 panic（close 已关闭的 channel 会 panic）。 +func TestStopIsIdempotent(t *testing.T) { + p := New("agentcli") + sdkInst := sdk.New("agentcli", sdk.SDKConfig{ + RegTool: func(string, sdk.ToolDef, sdk.ToolHandler) error { return nil }, + RegStage: func(sdk.Stage, sdk.StageHandler) {}, + RegAPI: func(string) error { return nil }, + Settings: sdk.NewSettings("agentcli", nil), + }) + sdkInst.SetIOInjector(&injectCapture{}) + if err := p.Start(sdkInst); err != nil { + t.Fatal(err) + } + defer func() { + if r := recover(); r != nil { + t.Fatalf("Stop() panicked on repeated call: %v", r) + } + }() + if err := p.Stop(); err != nil { + t.Fatal(err) + } + if err := p.Stop(); err != nil { + t.Fatal(err) + } +} diff --git a/internal/plugins/agentcli/pty_linux.go b/internal/plugins/agentcli/pty_linux.go index 844f707..477d4ae 100644 --- a/internal/plugins/agentcli/pty_linux.go +++ b/internal/plugins/agentcli/pty_linux.go @@ -6,6 +6,7 @@ import ( "fmt" "os" "os/exec" + "sync" "syscall" "unsafe" ) @@ -69,6 +70,12 @@ type linuxPty struct { master *os.File slave *os.File cmd *exec.Cmd + + // waitOnce/waitErr 保证 cmd.Wait() 恰好被调一次（重复 Wait 会报错）。 + // 为什么必须 Wait：newCommandPty 只 Start 不 Wait，子进程退出后没人回收， + // 内核里会积下一堆 僵尸（线上实测 homed 名下已有一个）。 + waitOnce sync.Once + waitErr error } func (p *linuxPty) Read(buf []byte) (int, error) { return p.master.Read(buf) } @@ -88,12 +95,31 @@ func (p *linuxPty) Resize(rows, cols uint16) error { func (p *linuxPty) Running() bool { return true } func (p *linuxPty) Kill() error { - if p.cmd != nil && p.cmd.Process != nil { + if p.cmd == nil || p.cmd.Process == nil { + return nil + } + // 杀整个进程组，而不是只杀直接子进程。 + // + // newCommandPty 用了 Setsid，所以 sh 是新 session/pgid 的领头进程， + // pgid == sh 的 pid；命令真正的进程（如 sleep、vim）是它的子进程， + // 同属这个 pgid。只 Kill(sh) 会留下孤儿 sleep 继续跑（实测：终端超时 + // 后 `sleep 300` 仍在，只是被 init 收养）。给负 pid 发信号 = 杀全组。 + pid := p.cmd.Process.Pid + if err := syscall.Kill(-pid, syscall.SIGKILL); err != nil { + // 进程组不可用（已消亡/无权限）时回落到单进程 Kill。 return p.cmd.Process.Kill() } return nil } +// Wait 回收子进程，避免 僵尸堆积。 +func (p *linuxPty) Wait() error { + p.waitOnce.Do(func() { + p.waitErr = p.cmd.Wait() + }) + return p.waitErr +} + func (p *linuxPty) Close() error { p.slave.Close() return p.master.Close() @@ -130,4 +156,4 @@ func newCommandPty(command string, rows, cols uint16) (ptyTerm, *exec.Cmd, error return nil, nil, fmt.Errorf("resize pty: %w", err) } return pt, cmd, nil -} \ No newline at end of file +}