From d62430a71bf1326b90dfa512fb215e54a1e39174 Mon Sep 17 00:00:00 2001 From: dev Date: Wed, 2 Sep 2026 10:59:59 +0800 Subject: [PATCH] =?UTF-8?q?feat(proc):=20=E5=AD=90=E8=BF=9B=E7=A8=8B?= =?UTF-8?q?=E9=80=9A=E9=81=93=20=E2=80=94=E2=80=94=20RPC=20=E5=8D=8F?= =?UTF-8?q?=E8=AE=AE=20+=20=E8=BF=9B=E7=A8=8B=E7=AE=A1=E7=90=86=EF=BC=88Pa?= =?UTF-8?q?rt=202=20=E6=A0=B8=E5=BF=83=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 协议面（protocol.go，§3.2 method id 平移为 method 名）: - NDJSON 帧，双向复用同一对 stdio；ID>0 需应答，ID==0 为通知(post-and-forget) - 51 个 C ABI method id 全部平移为可读 method 名并标注原编号对照 编号本身扔掉——加能力不用改两边常量表，不再有 47 夹在 7 和 8 之间的痕迹 - case 25(CORE_FREE_STRING) 无对应 method：进程模型下各自 GC，概念消失 - case 23/24(事件订阅) 与 io.setToolBlocks 今日均为空实现「给不了」， 子进程下首次真正可给（§3.8 能力对齐） - 新增 stage.lock/stage.unlock（C ABI 下不存在跨进程锁概念） - StageInvokeParams 不含 StageContext 数据本身——数据在共享段，只带 stage 名 + seq 进程面（process.go，§2.3 保留现有生命周期机制）: - Spawn: 启动 + 握手（协议版本不匹配显式拒绝，不半兼容运行） - readLoop: NDJSON 分派应答/插件反向请求，1MB 单帧上限（大 payload 走 arena） - CallContext: ctx 取消时立即返回**且清理 pending 条目** 对比 cgo：超时只让调用方返回，goroutine 永久卡在 C 调用里（现网泄漏 26 次） - Notify: ID=0 不占 pending 表，满足约束 B（流式逐 token 发布不得等待消费者） - markExited: EOF/退出 → 唤醒全部在途调用 → onExit 回调 这是「把 panic 捕获换成进程退出检测」的落点，plugin_health 逻辑完全复用 - Stop: plugin.stop → 宽限期 → 超时 Kill；Kill 后 OS 回收全部资源，零泄漏 - serveRequest 带 panic 隔离：内核 handler panic 不带崩 readLoop 验证（10 项，真实子进程而非 mock，含 -race）: - 握手/工具调用/错误上报（插件失败调用方收到 error，非假成功） - 插件反向调用内核（tool.register + settings.get 双向往返） - **崩溃隔离**：插件 panic → 子进程 exit 2，内核存活、收到 onExit、在途调用不挂死 - 优雅停止 / **Kill 卡死插件**（ctx 超时返回 + pending 清零 + 资源回收） - 通知不等应答（100 条 < 1s）/ 50 并发调用应答不串 / 协议版本不匹配拒绝 接口冻结: git diff third_party/homeagent-sdk/sdk/ 为空 --- internal/plugin/proc/process.go | 499 ++++++++++++++++++ internal/plugin/proc/process_test.go | 334 ++++++++++++ internal/plugin/proc/protocol.go | 215 ++++++++ .../plugin/proc/testdata/badprotoplugin.go | 52 ++ .../plugin/proc/testdata/callbackplugin.go | 120 +++++ internal/plugin/proc/testdata/crashplugin.go | 54 ++ internal/plugin/proc/testdata/echoplugin.go | 76 +++ internal/plugin/proc/testdata/hangplugin.go | 56 ++ 8 files changed, 1406 insertions(+) create mode 100644 internal/plugin/proc/process.go create mode 100644 internal/plugin/proc/process_test.go create mode 100644 internal/plugin/proc/protocol.go create mode 100644 internal/plugin/proc/testdata/badprotoplugin.go create mode 100644 internal/plugin/proc/testdata/callbackplugin.go create mode 100644 internal/plugin/proc/testdata/crashplugin.go create mode 100644 internal/plugin/proc/testdata/echoplugin.go create mode 100644 internal/plugin/proc/testdata/hangplugin.go diff --git a/internal/plugin/proc/process.go b/internal/plugin/proc/process.go new file mode 100644 index 0000000..16e3102 --- /dev/null +++ b/internal/plugin/proc/process.go @@ -0,0 +1,499 @@ +package proc + +import ( + "bufio" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "log" + "os" + "os/exec" + "sync" + "sync/atomic" + "time" + + "gitcode.com/JianFeeeee/HomeAgent/internal/meta" +) + +// Process 管理一个外部插件子进程：spawn / 双向 JSON-RPC / 优雅停止 / 崩溃检测。 +// +// 设计依据：docs/zh/架构迁移评估.md §4.1 阶段 2、§2.3（保留现有生命周期机制） +// +// 与 C ABI 路径的关键差异： +// - **崩溃隔离**：插件 panic 只让子进程退出，homed 存活（今日 panic 跨 C 栈可带崩内核） +// - **真正的取消**：Kill() 后 OS 回收全部资源，零泄漏 +// （今日 cgo 调用不可抢占，超时后 OS 线程永久占用，现网已泄漏 26 次，§9.3） +// - **可同步等真实结果**：RPC 天然可等应答 +// （今日 cgo 不可嵌套，output_send 只能异步、永远假成功，§9.4） +type Process struct { + name string + bin string + dir string + + cmd *exec.Cmd + stdin *bufio.Writer + stdout io.ReadCloser + + // writeMu 串行化 stdin 写入：NDJSON 帧不能交错，否则对端解析错乱。 + writeMu sync.Mutex + + // pending 表：请求 ID → 应答通道。 + mu sync.Mutex + nextID uint64 + pending map[uint64]chan *Response + closed bool + + // handler 处理插件反向发起的调用（51 个 core.* method）。 + handler RequestHandler + + // exited 在 readLoop 检测到 EOF/进程退出后关闭，用于唤醒所有等待者。 + exited chan struct{} + exitOnce sync.Once + exitErr atomic.Pointer[error] + readerWG sync.WaitGroup + readyOnce sync.Once + ready chan struct{} + + // onExit 在进程退出时回调（内核用它喂 plugin_health.recordCrash， + // 以及 ForceRelease 释放该插件持有的 stage 锁）。 + onExit func(name string, err error) + + // shmSize 是握手时告知插件的共享段大小（0 表示本插件不用共享段）。 + shmSize int +} + +// RequestHandler 处理插件 → 内核的调用。 +// 返回值会被序列化为 Response.Result；返回 error 则序列化为 Response.Error。 +type RequestHandler func(method string, params json.RawMessage) (interface{}, error) + +// Options 是 Spawn 的可选配置。 +type Options struct { + // Dir 是子进程工作目录（通常为插件目录）。 + Dir string + // Env 追加到子进程环境变量。 + Env []string + // ExtraFiles 传给子进程的额外文件描述符（fd 3 起）。 + // 共享内存段的 memfd 经此传递——子进程 mmap fd 3 即挂载同一段。 + ExtraFiles []*os.File + // ShmSize 是共享段大小，握手时告知插件（与 ExtraFiles[0] 的 memfd 对应）。 + ShmSize int + // Handler 处理插件反向调用。 + Handler RequestHandler + // OnExit 进程退出回调。 + OnExit func(name string, err error) + // HandshakeTimeout 建链超时，默认 10s。 + HandshakeTimeout time.Duration +} + +// 默认超时。 +const ( + defaultHandshakeTimeout = 10 * time.Second + // stopGracePeriod 是发出 plugin.stop 后等待进程自行退出的时间。 + // 超时则 Kill——**这是"真正的取消"**，对比 cgo 路径超时后线程永久泄漏。 + stopGracePeriod = 5 * time.Second +) + +// ErrProcessExited 表示子进程已退出，调用无法完成。 +var ErrProcessExited = errors.New("proc: 插件进程已退出") + +// Spawn 启动插件子进程并完成握手。 +func Spawn(name, bin string, opts Options) (*Process, error) { + if opts.Handler == nil { + return nil, fmt.Errorf("proc: %s 缺少 RequestHandler（插件无法回调内核）", name) + } + timeout := opts.HandshakeTimeout + if timeout <= 0 { + timeout = defaultHandshakeTimeout + } + + cmd := exec.Command(bin) + cmd.Dir = opts.Dir + // stderr 直通内核日志：插件的 panic 栈、log 输出可直接看到。 + cmd.Stderr = os.Stderr + if len(opts.Env) > 0 { + cmd.Env = append(os.Environ(), opts.Env...) + } + cmd.ExtraFiles = opts.ExtraFiles + + stdinPipe, err := cmd.StdinPipe() + if err != nil { + return nil, fmt.Errorf("proc: %s stdin 管道: %w", name, err) + } + stdoutPipe, err := cmd.StdoutPipe() + if err != nil { + return nil, fmt.Errorf("proc: %s stdout 管道: %w", name, err) + } + + p := &Process{ + name: name, + bin: bin, + dir: opts.Dir, + cmd: cmd, + stdin: bufio.NewWriter(stdinPipe), + stdout: stdoutPipe, + pending: make(map[uint64]chan *Response), + handler: opts.Handler, + exited: make(chan struct{}), + ready: make(chan struct{}), + onExit: opts.OnExit, + shmSize: opts.ShmSize, + } + + if err := cmd.Start(); err != nil { + return nil, fmt.Errorf("proc: 启动 %s (%s): %w", name, bin, err) + } + + p.readerWG.Add(1) + go p.readLoop() + + // 等 readLoop 就绪后再握手，避免应答早于 reader 启动而丢失。 + <-p.ready + + if err := p.handshake(timeout); err != nil { + p.Kill() + return nil, err + } + return p, nil +} + +// Name 返回插件名。 +func (p *Process) Name() string { return p.name } + +// PID 返回子进程 PID（用于诊断/日志）。 +func (p *Process) PID() int { + if p.cmd == nil || p.cmd.Process == nil { + return 0 + } + return p.cmd.Process.Pid +} + +// Exited 返回一个在进程退出时关闭的通道。 +func (p *Process) Exited() <-chan struct{} { return p.exited } + +// ExitError 返回进程退出原因（正常退出为 nil）。 +func (p *Process) ExitError() error { + if e := p.exitErr.Load(); e != nil { + return *e + } + return nil +} + +func (p *Process) handshake(timeout time.Duration) error { + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + + raw, err := p.CallContext(ctx, MethodHandshake, HandshakeParams{ + Protocol: ProtocolVersion, + CoreVersion: meta.Version, + PluginName: p.name, + ShmVersion: shmVersion, + ShmSize: p.shmSize, + }) + if err != nil { + return fmt.Errorf("proc: %s 握手失败: %w", p.name, err) + } + var res HandshakeResult + if err := json.Unmarshal(raw, &res); err != nil { + return fmt.Errorf("proc: %s 握手应答解析失败: %w", p.name, err) + } + if res.Protocol != ProtocolVersion { + return fmt.Errorf("proc: %s 协议版本不匹配（插件 %d，内核 %d）——请用配套 plugindev 重编", + p.name, res.Protocol, ProtocolVersion) + } + log.Printf("[proc] %s 已建链（pid=%d protocol=%d sdk=%s）", + p.name, p.PID(), res.Protocol, res.SDKVersion) + return nil +} + +// readLoop 读取子进程 stdout 的 NDJSON 帧，分派为「应答」或「插件发起的请求」。 +// +// 参考 clawhubadapter/sidecarProcess 的成熟做法：大 buffer 防长行截断、 +// pending 表定位应答、退出时唤醒全部等待者。 +func (p *Process) readLoop() { + defer p.readerWG.Done() + + scanner := bufio.NewScanner(bufio.NewReader(p.stdout)) + // 单帧上限 1MB：控制面帧本应很小（工具结果中位 93B）， + // 超大 payload 应走共享内存 arena 而非 RPC 帧。 + scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) + + p.readyOnce.Do(func() { close(p.ready) }) + + for scanner.Scan() { + line := scanner.Bytes() + if len(line) == 0 { + continue + } + // 帧可能是 Response（有 id 无 method）或 Request（有 method）。 + var probe struct { + ID uint64 `json:"id"` + Method string `json:"method"` + } + if err := json.Unmarshal(line, &probe); err != nil { + log.Printf("[proc] %s 收到非法 JSON 帧（%d 字节）: %v", p.name, len(line), err) + continue + } + + if probe.Method != "" { + // 插件发起的调用：拷贝一份再交给 goroutine（scanner 会复用底层数组） + buf := make([]byte, len(line)) + copy(buf, line) + go p.serveRequest(buf) + continue + } + + var resp Response + if err := json.Unmarshal(line, &resp); err != nil { + log.Printf("[proc] %s 应答解析失败: %v", p.name, err) + continue + } + p.mu.Lock() + ch, ok := p.pending[resp.ID] + delete(p.pending, resp.ID) + p.mu.Unlock() + if !ok { + log.Printf("[proc] %s 收到未知 id=%d 的应答（可能已超时）", p.name, resp.ID) + continue + } + ch <- &resp + } + + if err := scanner.Err(); err != nil { + log.Printf("[proc] %s 读取 stdout 出错: %v", p.name, err) + } + + // stdout 关闭（EOF）意味着进程结束——2.5ms 内即可感知（实验 6）。 + p.markExited() +} + +// markExited 回收进程、唤醒所有等待者、触发 onExit 回调。 +// +// 这是「把 panic 捕获换成进程退出检测」的落点（§2.3）： +// plugin_health 的 recordCrash / 冷却 / 自愈 / pendingReloads 全部逻辑复用， +// 只是信号源从 recover() 变成进程退出。 +func (p *Process) markExited() { + p.exitOnce.Do(func() { + waitErr := p.cmd.Wait() + if waitErr != nil { + e := fmt.Errorf("插件进程 %s 异常退出: %w", p.name, waitErr) + p.exitErr.Store(&e) + log.Printf("[proc] %s 退出: %v", p.name, waitErr) + } else { + log.Printf("[proc] %s 正常退出", p.name) + } + + p.mu.Lock() + p.closed = true + waiters := make([]chan *Response, 0, len(p.pending)) + for id, ch := range p.pending { + waiters = append(waiters, ch) + delete(p.pending, id) + } + p.mu.Unlock() + + // 唤醒所有在途调用，避免调用方挂死到自己的超时 + for _, ch := range waiters { + ch <- &Response{Error: ErrProcessExited.Error()} + } + + close(p.exited) + if p.onExit != nil { + p.onExit(p.name, p.ExitError()) + } + }) +} + +// serveRequest 处理插件反向发起的调用。 +func (p *Process) serveRequest(line []byte) { + var req Request + if err := json.Unmarshal(line, &req); err != nil { + log.Printf("[proc] %s 请求解析失败: %v", p.name, err) + return + } + + // panic 隔离：插件的回调参数可能触发内核 handler 的 panic， + // 不能让它带崩整个 readLoop（更不能带崩 homed）。 + var ( + result interface{} + err error + ) + func() { + defer func() { + if r := recover(); r != nil { + err = fmt.Errorf("内核 handler 处理 %s 时 panic: %v", req.Method, r) + log.Printf("[proc] %s: %v", p.name, err) + } + }() + result, err = p.handler(req.Method, req.Params) + }() + + // ID==0 是通知，不回应答（§2.4 约束 B：post-and-forget） + if req.ID == 0 { + if err != nil { + log.Printf("[proc] %s 通知 %s 处理失败: %v", p.name, req.Method, err) + } + return + } + + resp := Response{ID: req.ID} + if err != nil { + resp.Error = err.Error() + } else if result != nil { + if b, mErr := json.Marshal(result); mErr == nil { + resp.Result = b + } else { + resp.Error = fmt.Sprintf("结果序列化失败: %v", mErr) + } + } + if wErr := p.writeFrame(&resp); wErr != nil { + log.Printf("[proc] %s 回写应答失败: %v", p.name, wErr) + } +} + +// writeFrame 序列化并写入一帧（串行化，NDJSON 不能交错）。 +func (p *Process) writeFrame(v interface{}) error { + b, err := json.Marshal(v) + if err != nil { + return err + } + p.writeMu.Lock() + defer p.writeMu.Unlock() + if _, err := p.stdin.Write(b); err != nil { + return err + } + if err := p.stdin.WriteByte('\n'); err != nil { + return err + } + return p.stdin.Flush() +} + +// Call 发起 RPC 并等待应答（无超时上限，由调用方 context 控制）。 +func (p *Process) Call(method string, params interface{}) (json.RawMessage, error) { + return p.CallContext(context.Background(), method, params) +} + +// CallContext 发起 RPC 并等待应答，受 ctx 取消/超时控制。 +// +// **ctx 取消时调用方立即返回，且 pending 条目被清理**—— +// 对比 cgo 路径：超时只让调用方返回，goroutine 仍永久卡在 C 调用里（§9.3）。 +// 这里子进程若真卡住，上层可 Kill()，OS 回收全部资源。 +func (p *Process) CallContext(ctx context.Context, method string, params interface{}) (json.RawMessage, error) { + var raw json.RawMessage + if params != nil { + b, err := json.Marshal(params) + if err != nil { + return nil, fmt.Errorf("proc: %s 序列化 %s 参数: %w", p.name, method, err) + } + raw = b + } + + ch := make(chan *Response, 1) + + p.mu.Lock() + if p.closed { + p.mu.Unlock() + return nil, fmt.Errorf("proc: %s 调用 %s: %w", p.name, method, ErrProcessExited) + } + p.nextID++ + id := p.nextID + p.pending[id] = ch + p.mu.Unlock() + + if err := p.writeFrame(&Request{ID: id, Method: method, Params: raw}); err != nil { + p.mu.Lock() + delete(p.pending, id) + p.mu.Unlock() + return nil, fmt.Errorf("proc: %s 发送 %s: %w", p.name, method, err) + } + + select { + case resp := <-ch: + if resp.Error != "" { + return nil, fmt.Errorf("proc: %s.%s: %s", p.name, method, resp.Error) + } + return resp.Result, nil + case <-ctx.Done(): + p.mu.Lock() + delete(p.pending, id) + p.mu.Unlock() + return nil, fmt.Errorf("proc: %s 调用 %s: %w", p.name, method, ctx.Err()) + case <-p.exited: + return nil, fmt.Errorf("proc: %s 调用 %s: %w", p.name, method, ErrProcessExited) + } +} + +// Notify 发送不需要应答的通知（ID=0，fire-and-forget）。 +// +// 用于事件投递等路径：内核发通知**绝不等待消费者**（§2.4 约束 B—— +// 流式输出逐 token 发布，任何等待都会造成卡顿）。 +func (p *Process) Notify(method string, params interface{}) error { + var raw json.RawMessage + if params != nil { + b, err := json.Marshal(params) + if err != nil { + return err + } + raw = b + } + p.mu.Lock() + closed := p.closed + p.mu.Unlock() + if closed { + return ErrProcessExited + } + return p.writeFrame(&Request{Method: method, Params: raw}) +} + +// Stop 优雅停止：发 plugin.stop → 等宽限期 → 超时则 Kill。 +// +// 插件侧收到 plugin.stop 后应先跑 RunStopHandlers 再 Stop()， +// 与 C ABI 路径的停止链路语义一致（§2.3 已验证被正确调用）。 +func (p *Process) Stop() error { + select { + case <-p.exited: + return nil // 已经退出 + default: + } + + ctx, cancel := context.WithTimeout(context.Background(), stopGracePeriod) + defer cancel() + if _, err := p.CallContext(ctx, MethodPluginStop, nil); err != nil { + // 停止调用失败不影响后续 Kill——插件可能已经崩了 + if !errors.Is(err, ErrProcessExited) { + log.Printf("[proc] %s plugin.stop 失败（将强制结束）: %v", p.name, err) + } + } + + select { + case <-p.exited: + return nil + case <-time.After(stopGracePeriod): + log.Printf("[proc] %s 宽限期内未退出，强制结束", p.name) + return p.Kill() + } +} + +// Kill 强制结束子进程并回收资源。 +// +// **这是 C ABI 路径拿不到的能力**：cgo 调用不可被 Go runtime 抢占或取消， +// 超时后该 OS 线程永久占用（实验 14 实测 20 次调用线性泄漏 +18 线程）。 +// 子进程模型下 Kill 后 OS 回收全部资源，零泄漏。 +func (p *Process) Kill() error { + if p.cmd == nil || p.cmd.Process == nil { + return nil + } + err := p.cmd.Process.Kill() + // 等 readLoop 观察到 EOF 并完成 Wait/清理 + select { + case <-p.exited: + case <-time.After(2 * time.Second): + p.markExited() // 兜底：极端情况下强制走清理 + } + p.readerWG.Wait() + if err != nil && !errors.Is(err, os.ErrProcessDone) { + return fmt.Errorf("proc: 结束 %s: %w", p.name, err) + } + return nil +} diff --git a/internal/plugin/proc/process_test.go b/internal/plugin/proc/process_test.go new file mode 100644 index 0000000..50c2893 --- /dev/null +++ b/internal/plugin/proc/process_test.go @@ -0,0 +1,334 @@ +package proc + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "os" + "os/exec" + "path/filepath" + "strings" + "sync" + "testing" + "time" +) + +// Process 的测试用真实子进程（go build 出的小二进制），而非 mock： +// 崩溃隔离、EOF 感知、Kill 回收这些性质只有真进程才能验证—— +// 它们恰是迁移相对 C ABI 的核心收益（§9.5）。 + +// buildTestPlugin 编译 testdata 下的假插件，返回二进制路径。 +func buildTestPlugin(t *testing.T, srcName string) string { + t.Helper() + if _, err := exec.LookPath("go"); err != nil { + t.Skip("环境无 go 工具链，跳过子进程测试") + } + + src := filepath.Join("testdata", srcName) + if _, err := os.Stat(src); err != nil { + t.Fatalf("测试插件源码缺失 %s: %v", src, err) + } + + bin := filepath.Join(t.TempDir(), strings.TrimSuffix(srcName, ".go")) + cmd := exec.Command("go", "build", "-o", bin, src) + cmd.Env = append(os.Environ(), "CGO_ENABLED=0") + if out, err := cmd.CombinedOutput(); err != nil { + t.Fatalf("编译测试插件 %s 失败: %v\n%s", srcName, err, out) + } + return bin +} + +// noopHandler 是最简的内核侧 handler（测试中不需要真实 core.* 能力）。 +func noopHandler(method string, params json.RawMessage) (interface{}, error) { + return nil, fmt.Errorf("测试环境未实现 %s", method) +} + +func TestProcess_SpawnHandshakeAndToolInvoke(t *testing.T) { + bin := buildTestPlugin(t, "echoplugin.go") + + p, err := Spawn("echo", bin, Options{Handler: noopHandler}) + if err != nil { + t.Fatalf("Spawn: %v", err) + } + defer p.Kill() + + if p.PID() == 0 { + t.Error("PID 应非零") + } + + raw, err := p.Call(MethodToolInvoke, ToolInvokeParams{ + Name: "echo_tool", + Args: map[string]interface{}{"text": "你好"}, + }) + if err != nil { + t.Fatalf("tool.invoke: %v", err) + } + var res ToolInvokeResult + if err := json.Unmarshal(raw, &res); err != nil { + t.Fatalf("解析应答: %v", err) + } + if res.Result != "你好" { + t.Fatalf("工具应回显 '你好'，实际 %v", res.Result) + } +} + +// 插件返回错误时调用方必须收到 error —— 对比 C ABI 路径的 output_send +// 永远返回成功（§9.4，现网 2 次消息发不出而模型以为成功）。 +func TestProcess_PluginErrorIsReported(t *testing.T) { + bin := buildTestPlugin(t, "echoplugin.go") + p, err := Spawn("echo", bin, Options{Handler: noopHandler}) + if err != nil { + t.Fatalf("Spawn: %v", err) + } + defer p.Kill() + + _, err = p.Call(MethodToolInvoke, ToolInvokeParams{Name: "fail_tool"}) + if err == nil { + t.Fatal("插件返回错误时调用方应收到 error") + } + if !strings.Contains(err.Error(), "故意失败") { + t.Errorf("错误信息应透传插件的原因，实际: %v", err) + } +} + +// 插件反向调用内核（51 个 core.* method 的机制验证）。 +func TestProcess_PluginCallsBackIntoKernel(t *testing.T) { + bin := buildTestPlugin(t, "callbackplugin.go") + + var ( + mu sync.Mutex + gotCall []string + ) + handler := func(method string, params json.RawMessage) (interface{}, error) { + mu.Lock() + gotCall = append(gotCall, method) + mu.Unlock() + switch method { + case MethodSettingsGet: + return map[string]interface{}{"value": "配置值"}, nil + case MethodToolRegister: + return nil, nil + } + return nil, fmt.Errorf("未实现 %s", method) + } + + p, err := Spawn("cb", bin, Options{Handler: handler}) + if err != nil { + t.Fatalf("Spawn: %v", err) + } + defer p.Kill() + + // plugin.start 期间插件会回调 tool.register + settings.get + if _, err := p.Call(MethodPluginStart, nil); err != nil { + t.Fatalf("plugin.start: %v", err) + } + + mu.Lock() + defer mu.Unlock() + if len(gotCall) < 2 { + t.Fatalf("内核应收到插件的反向调用，实际 %v", gotCall) + } + hasRegister, hasSettings := false, false + for _, m := range gotCall { + if m == MethodToolRegister { + hasRegister = true + } + if m == MethodSettingsGet { + hasSettings = true + } + } + if !hasRegister || !hasSettings { + t.Errorf("应收到 tool.register 与 settings.get，实际 %v", gotCall) + } +} + +// 崩溃隔离：插件 panic 只让子进程退出，内核存活并收到 onExit（§9.5 表格第 2 行）。 +// C ABI 路径下 panic 跨 C 栈，recover 兜不住会带崩整个 homed（§1.4）。 +func TestProcess_CrashIsolationAndExitDetection(t *testing.T) { + bin := buildTestPlugin(t, "crashplugin.go") + + exitCh := make(chan error, 1) + p, err := Spawn("crash", bin, Options{ + Handler: noopHandler, + OnExit: func(name string, err error) { exitCh <- err }, + }) + if err != nil { + t.Fatalf("Spawn: %v", err) + } + + // 触发插件 panic + _, callErr := p.Call(MethodToolInvoke, ToolInvokeParams{Name: "boom"}) + if callErr == nil { + t.Error("插件崩溃时在途调用应返回错误，而非挂死") + } + + select { + case exitErr := <-exitCh: + if exitErr == nil { + t.Error("panic 退出应报告非 nil 错误（供 recordCrash 使用）") + } + case <-time.After(5 * time.Second): + t.Fatal("未在 5s 内检测到进程退出（EOF 感知失效）") + } + + select { + case <-p.Exited(): + case <-time.After(time.Second): + t.Error("Exited() 通道应已关闭") + } + + // 进程已退出后继续调用应立即失败，不能挂死 + if _, err := p.Call(MethodToolInvoke, ToolInvokeParams{Name: "echo_tool"}); !errors.Is(err, ErrProcessExited) { + t.Errorf("退出后调用应返回 ErrProcessExited，实际 %v", err) + } +} + +// 优雅停止：plugin.stop 后进程自行退出。 +func TestProcess_GracefulStop(t *testing.T) { + bin := buildTestPlugin(t, "echoplugin.go") + p, err := Spawn("echo", bin, Options{Handler: noopHandler}) + if err != nil { + t.Fatalf("Spawn: %v", err) + } + + if err := p.Stop(); err != nil { + t.Fatalf("Stop: %v", err) + } + select { + case <-p.Exited(): + case <-time.After(3 * time.Second): + t.Fatal("Stop 后进程应退出") + } + if err := p.ExitError(); err != nil { + t.Errorf("优雅停止应无错误退出，实际 %v", err) + } +} + +// **真正的取消**：卡死的插件可被 Kill 回收（§9.3 对照 cgo 超时线程永久泄漏）。 +func TestProcess_KillHungPlugin(t *testing.T) { + bin := buildTestPlugin(t, "hangplugin.go") + p, err := Spawn("hang", bin, Options{Handler: noopHandler}) + if err != nil { + t.Fatalf("Spawn: %v", err) + } + + // 调用会卡住，用 context 超时返回（调用方不被拖死） + ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond) + defer cancel() + _, err = p.CallContext(ctx, MethodToolInvoke, ToolInvokeParams{Name: "hang_tool"}) + if err == nil { + t.Fatal("卡死的调用应因 ctx 超时返回") + } + if !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("应为 DeadlineExceeded，实际 %v", err) + } + + // pending 条目必须已清理（不泄漏） + p.mu.Lock() + pendingCount := len(p.pending) + p.mu.Unlock() + if pendingCount != 0 { + t.Errorf("超时后 pending 表应清空，实际残留 %d 条", pendingCount) + } + + // Kill 真正回收资源 + if err := p.Kill(); err != nil { + t.Fatalf("Kill: %v", err) + } + select { + case <-p.Exited(): + case <-time.After(3 * time.Second): + t.Fatal("Kill 后进程应退出") + } +} + +// 通知（ID=0）不等应答——事件投递路径必须 post-and-forget（§2.4 约束 B）。 +func TestProcess_NotifyDoesNotWait(t *testing.T) { + bin := buildTestPlugin(t, "echoplugin.go") + p, err := Spawn("echo", bin, Options{Handler: noopHandler}) + if err != nil { + t.Fatalf("Spawn: %v", err) + } + defer p.Kill() + + start := time.Now() + for i := 0; i < 100; i++ { + if err := p.Notify("event.deliver", map[string]interface{}{"seq": i}); err != nil { + t.Fatalf("Notify: %v", err) + } + } + elapsed := time.Since(start) + // 100 条通知若每条都等应答，至少要 100 个往返；post-and-forget 应远快于此 + if elapsed > time.Second { + t.Errorf("100 条通知耗时 %v，疑似在等应答（应 post-and-forget）", elapsed) + } + + // 通知不占 pending 表 + p.mu.Lock() + pendingCount := len(p.pending) + p.mu.Unlock() + if pendingCount != 0 { + t.Errorf("通知不应占用 pending 表，实际 %d 条", pendingCount) + } +} + +// 并发调用：pending 表按 ID 正确路由，应答不串。 +func TestProcess_ConcurrentCallsRouteCorrectly(t *testing.T) { + bin := buildTestPlugin(t, "echoplugin.go") + p, err := Spawn("echo", bin, Options{Handler: noopHandler}) + if err != nil { + t.Fatalf("Spawn: %v", err) + } + defer p.Kill() + + const n = 50 + var wg sync.WaitGroup + errs := make(chan error, n) + for i := 0; i < n; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + want := fmt.Sprintf("msg-%d", i) + raw, err := p.Call(MethodToolInvoke, ToolInvokeParams{ + Name: "echo_tool", + Args: map[string]interface{}{"text": want}, + }) + if err != nil { + errs <- err + return + } + var res ToolInvokeResult + if err := json.Unmarshal(raw, &res); err != nil { + errs <- err + return + } + if res.Result != want { + errs <- fmt.Errorf("应答串了：期望 %q，实际 %v", want, res.Result) + } + }(i) + } + wg.Wait() + close(errs) + for err := range errs { + t.Errorf("并发调用失败: %v", err) + } +} + +// 协议版本不匹配必须显式拒绝，不能半兼容运行。 +func TestProcess_ProtocolMismatchRejected(t *testing.T) { + bin := buildTestPlugin(t, "badprotoplugin.go") + _, err := Spawn("badproto", bin, Options{Handler: noopHandler}) + if err == nil { + t.Fatal("协议版本不匹配应拒绝建链") + } + if !strings.Contains(err.Error(), "协议版本不匹配") { + t.Errorf("错误应说明版本不匹配，实际: %v", err) + } +} + +func TestProcess_SpawnRequiresHandler(t *testing.T) { + if _, err := Spawn("x", "/bin/true", Options{}); err == nil { + t.Fatal("缺少 Handler 应报错（插件无法回调内核）") + } +} diff --git a/internal/plugin/proc/protocol.go b/internal/plugin/proc/protocol.go new file mode 100644 index 0000000..74515a3 --- /dev/null +++ b/internal/plugin/proc/protocol.go @@ -0,0 +1,215 @@ +package proc + +import "encoding/json" + +// RPC 协议定义：控制面（§3.2 method id 平移为 method 名）。 +// +// 帧格式：**换行分隔的 JSON**（NDJSON），双向复用同一对 stdio 管道。 +// 内核 → 插件 stdin ：请求 / 响应 +// 插件 → 内核 stdout：请求 / 响应 +// +// 为什么不用 length-prefixed 二进制帧：工具调用结果中位数仅 93B（§2.5）， +// JSON 序列化 3-8 µs 对比 LLM 单轮 2-8 秒占 0.0001%，可读性与可调试性更值。 +// 大 payload（多媒体二进制）走共享内存 arena，不进 RPC 帧（§3.3 实验 10：18-22x）。 + +// 协议版本：与共享段版本独立演进。 +// 插件握手时上报，内核校验——不匹配显式拒绝，避免半兼容导致的诡异行为。 +const ProtocolVersion = 1 + +// Direction 无需显式字段：靠 Method 是否为空区分请求与响应 +// （与 clawhubadapter/sidecar 的成熟做法一致）。 + +// Request 是一次 RPC 调用。 +// +// ID 语义： +// - ID > 0 ：需要响应，调用方在 pending 表等待 +// - ID == 0 ：通知（fire-and-forget），被调方不得回响应 +// +// 通知用于事件投递等不关心结果的路径（§2.4 约束 B：内核发通知绝不等待消费者）。 +type Request struct { + ID uint64 `json:"id,omitempty"` + Method string `json:"method"` + Params json.RawMessage `json:"params,omitempty"` +} + +// Response 是对 Request 的应答。Error 非空表示失败。 +type Response struct { + ID uint64 `json:"id"` + Result json.RawMessage `json:"result,omitempty"` + Error string `json:"error,omitempty"` +} + +// ---- kernel → plugin（内核调用插件，对应今日 7 个 //export）---- +const ( + // MethodPluginInit 传插件名与配置，插件构造实例但不启动。 + MethodPluginInit = "plugin.init" + // MethodPluginStart 插件注册工具/阶段/通道（其间会反向发起大量 core.* 调用）。 + MethodPluginStart = "plugin.start" + // MethodPluginStop 优雅停止：插件侧先跑 RunStopHandlers 再 Stop()。 + MethodPluginStop = "plugin.stop" + // MethodToolInvoke 执行插件工具。 + MethodToolInvoke = "tool.invoke" + // MethodStageInvoke 执行阶段处理器。数据经共享段传递，参数只带阶段名与段世代号。 + MethodStageInvoke = "stage.invoke" + // MethodOutputInvoke 经插件输出通道发送。 + MethodOutputInvoke = "output.invoke" + // MethodHandshake 建链首帧：交换协议版本、SDK 版本、共享段规格。 + MethodHandshake = "handshake" +) + +// ---- plugin → kernel（51 个 method id 平移，§3.2）---- +// +// 编号本身扔掉：不再维护"下一个可用 id 是 52"，加能力不用改两边常量表， +// 也不再出现 47 夹在 7 和 8 之间的历史痕迹。 +const ( + // 注册面（原 case 1/2/3/4/46） + MethodToolRegister = "tool.register" // 1 CORE_REGISTER_TOOL + MethodStageRegister = "stage.register" // 2 CORE_REGISTER_STAGE + MethodOutputRegister = "output.register" // 3 CORE_REGISTER_OUTPUT_CH + MethodAPIRegister = "api.register" // 4 CORE_REGISTER_PLUGIN_API + MethodInputRegister = "input.register" // 46 CORE_REGISTER_INPUT_CH + + // IO 注入（原 case 5/6/7/47） + MethodIOInjectText = "io.injectText" // 5 CORE_INJECT_TEXT + MethodIOInjectInterrupt = "io.injectInterrupt" // 6 CORE_INJECT_INTERRUPT_TEXT + MethodIOInjectTextNoMem = "io.injectTextNoMem" // 7 CORE_INJECT_TEXT_NO_MEMORY + MethodIOInjectSync = "io.injectInputSync" // 47 CORE_INJECT_INPUT_SYNC + // MethodIOSetToolBlocks 多模态注入——今日 C ABI 侧是空实现（§1.4）， + // 子进程下二进制落 arena、描述符回传，首次真正可用。 + MethodIOSetToolBlocks = "io.setToolBlocks" + + // 生命周期（原 case 8） + MethodLifecycleAutoRestart = "lifecycle.autoRestart" // 8 CORE_SET_AUTO_RESTART + + // 图记忆（原 case 9/10/11/12/13） + MethodMemoryRecall = "memory.recall" // 9 + MethodMemoryCommit = "memory.commit" // 10 + MethodMemoryIntrospect = "memory.introspect" // 11 + MethodMemoryMerge = "memory.merge" // 12 + MethodMemoryPurge = "memory.purge" // 13 + + // 文档记忆（原 case 14/32/33/34） + MethodDocQuery = "doc.query" // 14 + MethodDocInsert = "doc.insert" // 32 + MethodDocRemove = "doc.remove" // 33 + MethodDocStats = "doc.stats" // 34 + + // 知识库（原 case 15/35/36） + MethodKnowledgeSearch = "knowledge.search" // 15 + MethodKnowledgeAdd = "knowledge.add" // 35 + MethodKnowledgeList = "knowledge.list" // 36 + + // 文本记忆（原 case 41） + MethodTextMemoryAppend = "textmemory.append" // 41 + + // 设置（原 case 16/17/18/26/27/28/29/30/31/42/43/44/45/51） + MethodSettingsGet = "settings.get" // 16 + MethodSettingsSet = "settings.set" // 17 + MethodSettingsRegisterDef = "settings.registerDef" // 18 + MethodSettingsGetCore = "settings.getCore" // 26 + MethodSettingsSetCore = "settings.setCore" // 27 + MethodSettingsListCore = "settings.listCore" // 28 + MethodSettingsGetPlugin = "settings.getPlugin" // 29 + MethodSettingsSetPlugin = "settings.setPlugin" // 30 + MethodSettingsListPlugin = "settings.listPlugin" // 31 + MethodSettingsList = "settings.list" // 42 + MethodSettingsDefs = "settings.defs" // 43 + MethodSettingsDump = "settings.dump" // 44 + MethodSettingsPlugins = "settings.plugins" // 45 + MethodSettingsDataDir = "settings.dataDir" // 51 + + // LLM 源（原 case 19/20/37） + MethodLLMListSources = "llm.listSources" // 19 + MethodLLMSetSource = "llm.setSource" // 20 + MethodLLMCurrentSource = "llm.currentSource" // 37 + + // 社交图（只读，原 case 21/22/38/39/40） + MethodSocialGetPerson = "social.getPerson" // 21 + MethodSocialGetNetwork = "social.getNetwork" // 22 + MethodSocialGetTrait = "social.getTrait" // 38 + MethodSocialGetRelation = "social.getRelations" // 39 + MethodSocialListPersons = "social.listPersons" // 40 + + // 事件（原 case 23/24 —— 今日均为空实现「给不了」， + // 子进程下经事件环 + eventfd 首次真正可用，见 §3.6/§3.8） + MethodEventsSubscribe = "events.subscribe" // 23 + MethodEventsUnsubscribe = "events.unsubscribe" // 24 + + // 插件管理（原 case 48/49/50） + MethodPluginReloadOne = "plugin.reloadOne" // 48 + MethodPluginListLoaded = "plugin.listLoaded" // 49 + MethodPluginIsDisabled = "plugin.isDisabled" // 50 + + // 共享段锁仲裁（新增，无对应 method id —— C ABI 下不存在跨进程锁概念） + MethodStageLock = "stage.lock" + MethodStageUnlock = "stage.unlock" +) + +// 原 case 25（CORE_FREE_STRING）无对应 RPC method： +// C ABI 下需要显式释放跨边界字符串，进程模型下由各自 GC 管理，概念消失。 + +// HandshakeParams 是内核 → 插件的建链首帧：告知内核侧规格。 +type HandshakeParams struct { + Protocol int `json:"protocol"` // 内核支持的协议版本 + CoreVersion string `json:"core_version"` // 内核版本（诊断用） + PluginName string `json:"plugin_name"` // 内核分配的插件名 + // ShmVersion 让插件确认共享段布局一致；不匹配时插件应拒绝启动而非错读。 + ShmVersion uint32 `json:"shm_version"` + // ShmSize 是内核分配的共享段大小，插件据此 mmap（段本身经 fd 3 传入）。 + ShmSize int `json:"shm_size"` +} + +// HandshakeResult 是插件 → 内核的建链应答：上报自身信息。 +type HandshakeResult struct { + Protocol int `json:"protocol"` // 必须等于 ProtocolVersion + SDKVersion string `json:"sdk_version"` // 插件编译时链接的公开 SDK 版本 + PluginName string `json:"plugin_name"` + PID int `json:"pid"` +} + +// StageInvokeParams 是 stage.invoke 的参数。 +// +// **注意：不含 StageContext 数据本身**——数据在共享段，此处只带定位信息。 +// 这是共享内存数据面的意义：并发改写同一份状态，而非各持副本 +// （副本模型实测 35.8~36.8% lost update，§8.4）。 +type StageInvokeParams struct { + Stage string `json:"stage"` + // Seq 是内核写入共享段后的世代号，插件读到的 seq 应 >= 此值。 + Seq uint64 `json:"seq"` +} + +// StageInvokeResult 是插件执行 stage 后的应答。 +type StageInvokeResult struct { + // DirtyFields 是插件实际写回共享段的字段数，0 表示只读插件。 + // 内核据此判断是否需要重读共享段，也用于诊断"谁改了什么"。 + DirtyFields int `json:"dirty_fields"` + // Seq 是插件写回后的世代号。 + Seq uint64 `json:"seq"` +} + +// ToolInvokeParams / ToolInvokeResult：工具调用（原 go_invoke_tool）。 +type ToolInvokeParams struct { + Name string `json:"name"` + Args map[string]interface{} `json:"args,omitempty"` +} + +type ToolInvokeResult struct { + Result interface{} `json:"result,omitempty"` +} + +// OutputInvokeParams：输出通道发送（原 go_invoke_output）。 +// +// 与 C ABI 路径的关键差异：**可同步等待真实结果**。 +// C ABI 下因 cgo 不可嵌套，只能异步 fire-and-forget，导致 output_send +// 永远返回成功（§9.4，现网 2 次消息发不出而模型以为成功）。 +// 进程模型下 RPC 天然可等应答，该缺陷从根上消失。 +type OutputInvokeParams struct { + Channel string `json:"channel"` + Args map[string]interface{} `json:"args,omitempty"` +} + +// PluginInitParams：插件构造参数（原 case init_plugin）。 +type PluginInitParams struct { + Name string `json:"name"` + Config map[string]interface{} `json:"config,omitempty"` +} diff --git a/internal/plugin/proc/testdata/badprotoplugin.go b/internal/plugin/proc/testdata/badprotoplugin.go new file mode 100644 index 0000000..a119d8b --- /dev/null +++ b/internal/plugin/proc/testdata/badprotoplugin.go @@ -0,0 +1,52 @@ +//go:build ignore + +// badprotoplugin 上报错误的协议版本，验证内核显式拒绝而非半兼容运行。 +package main + +import ( + "bufio" + "encoding/json" + "os" +) + +type request struct { + ID uint64 `json:"id,omitempty"` + Method string `json:"method"` + Params json.RawMessage `json:"params,omitempty"` +} + +type response struct { + ID uint64 `json:"id"` + Result interface{} `json:"result,omitempty"` + Error string `json:"error,omitempty"` +} + +func main() { + in := bufio.NewScanner(bufio.NewReader(os.Stdin)) + out := bufio.NewWriter(os.Stdout) + send := func(v interface{}) { + b, _ := json.Marshal(v) + out.Write(b) + out.WriteByte('\n') + out.Flush() + } + + for in.Scan() { + var req request + if err := json.Unmarshal(in.Bytes(), &req); err != nil { + continue + } + if req.Method == "handshake" { + send(response{ID: req.ID, Result: map[string]interface{}{ + "protocol": 999, // 故意不匹配 + "sdk_version": "ancient", + "plugin_name": "badproto", + "pid": os.Getpid(), + }}) + continue + } + if req.ID != 0 { + send(response{ID: req.ID}) + } + } +} diff --git a/internal/plugin/proc/testdata/callbackplugin.go b/internal/plugin/proc/testdata/callbackplugin.go new file mode 100644 index 0000000..9945323 --- /dev/null +++ b/internal/plugin/proc/testdata/callbackplugin.go @@ -0,0 +1,120 @@ +//go:build ignore + +// callbackplugin 验证插件 → 内核的反向调用（51 个 core.* method 的机制）。 +package main + +import ( + "bufio" + "encoding/json" + "os" + "sync" +) + +type request struct { + ID uint64 `json:"id,omitempty"` + Method string `json:"method"` + Params json.RawMessage `json:"params,omitempty"` +} + +type response struct { + ID uint64 `json:"id"` + Result interface{} `json:"result,omitempty"` + Error string `json:"error,omitempty"` +} + +var ( + out = bufio.NewWriter(os.Stdout) + writeMu sync.Mutex + nextID uint64 + pending = map[uint64]chan json.RawMessage{} + pendMu sync.Mutex +) + +func send(v interface{}) { + b, _ := json.Marshal(v) + writeMu.Lock() + out.Write(b) + out.WriteByte('\n') + out.Flush() + writeMu.Unlock() +} + +// callKernel 反向调用内核并等待应答。 +func callKernel(method string, params interface{}) json.RawMessage { + pendMu.Lock() + nextID++ + id := nextID + ch := make(chan json.RawMessage, 1) + pending[id] = ch + pendMu.Unlock() + + var raw json.RawMessage + if params != nil { + b, _ := json.Marshal(params) + raw = b + } + send(request{ID: id, Method: method, Params: raw}) + return <-ch +} + +func main() { + in := bufio.NewScanner(bufio.NewReader(os.Stdin)) + in.Buffer(make([]byte, 0, 64*1024), 1024*1024) + + for in.Scan() { + line := make([]byte, len(in.Bytes())) + copy(line, in.Bytes()) + + var probe struct { + ID uint64 `json:"id"` + Method string `json:"method"` + } + if err := json.Unmarshal(line, &probe); err != nil { + continue + } + + // 内核对我们反向调用的应答 + if probe.Method == "" { + var resp struct { + ID uint64 `json:"id"` + Result json.RawMessage `json:"result"` + } + json.Unmarshal(line, &resp) + pendMu.Lock() + ch, ok := pending[resp.ID] + delete(pending, resp.ID) + pendMu.Unlock() + if ok { + ch <- resp.Result + } + continue + } + + var req request + json.Unmarshal(line, &req) + switch req.Method { + case "handshake": + send(response{ID: req.ID, Result: map[string]interface{}{ + "protocol": 1, + "sdk_version": "test", + "plugin_name": "cb", + "pid": os.Getpid(), + }}) + case "plugin.start": + // 在独立 goroutine 里回调，避免阻塞读循环 + go func(id uint64) { + callKernel("tool.register", map[string]interface{}{"name": "cb_tool"}) + callKernel("settings.get", map[string]interface{}{"key": "some_key"}) + send(response{ID: id}) + }(req.ID) + case "plugin.stop": + send(response{ID: req.ID}) + out.Flush() + os.Exit(0) + default: + if req.ID != 0 { + send(response{ID: req.ID}) + } + } + } +} diff --git a/internal/plugin/proc/testdata/crashplugin.go b/internal/plugin/proc/testdata/crashplugin.go new file mode 100644 index 0000000..0bd87f9 --- /dev/null +++ b/internal/plugin/proc/testdata/crashplugin.go @@ -0,0 +1,54 @@ +//go:build ignore + +// crashplugin 在收到 boom 工具调用时 panic，用于验证崩溃隔离： +// 子进程死亡不应带崩 homed，且内核须能感知退出（供 recordCrash 使用）。 +package main + +import ( + "bufio" + "encoding/json" + "os" +) + +type request struct { + ID uint64 `json:"id,omitempty"` + Method string `json:"method"` + Params json.RawMessage `json:"params,omitempty"` +} + +type response struct { + ID uint64 `json:"id"` + Result interface{} `json:"result,omitempty"` + Error string `json:"error,omitempty"` +} + +func main() { + in := bufio.NewScanner(bufio.NewReader(os.Stdin)) + out := bufio.NewWriter(os.Stdout) + send := func(v interface{}) { + b, _ := json.Marshal(v) + out.Write(b) + out.WriteByte('\n') + out.Flush() + } + + for in.Scan() { + var req request + if err := json.Unmarshal(in.Bytes(), &req); err != nil { + continue + } + switch req.Method { + case "handshake": + send(response{ID: req.ID, Result: map[string]interface{}{ + "protocol": 1, "sdk_version": "test", "plugin_name": "crash", "pid": os.Getpid(), + }}) + case "tool.invoke": + // 模拟插件 bug：直接 panic，进程带非零码退出 + panic("插件内部 panic：用于验证崩溃隔离") + default: + if req.ID != 0 { + send(response{ID: req.ID}) + } + } + } +} diff --git a/internal/plugin/proc/testdata/echoplugin.go b/internal/plugin/proc/testdata/echoplugin.go new file mode 100644 index 0000000..1437a03 --- /dev/null +++ b/internal/plugin/proc/testdata/echoplugin.go @@ -0,0 +1,76 @@ +//go:build ignore + +// echoplugin 是测试用的最简子进程插件：实现握手 + 回显工具。 +// 不 import 公开 SDK——只验证 proc 包的 RPC 机制本身。 +package main + +import ( + "bufio" + "encoding/json" + "fmt" + "os" +) + +type request struct { + ID uint64 `json:"id,omitempty"` + Method string `json:"method"` + Params json.RawMessage `json:"params,omitempty"` +} + +type response struct { + ID uint64 `json:"id"` + Result interface{} `json:"result,omitempty"` + Error string `json:"error,omitempty"` +} + +func main() { + in := bufio.NewScanner(bufio.NewReader(os.Stdin)) + in.Buffer(make([]byte, 0, 64*1024), 1024*1024) + out := bufio.NewWriter(os.Stdout) + + send := func(v interface{}) { + b, _ := json.Marshal(v) + out.Write(b) + out.WriteByte('\n') + out.Flush() + } + + for in.Scan() { + var req request + if err := json.Unmarshal(in.Bytes(), &req); err != nil { + continue + } + switch req.Method { + case "handshake": + send(response{ID: req.ID, Result: map[string]interface{}{ + "protocol": 1, + "sdk_version": "test", + "plugin_name": "echo", + "pid": os.Getpid(), + }}) + case "plugin.stop": + send(response{ID: req.ID}) + out.Flush() + os.Exit(0) + case "tool.invoke": + var p struct { + Name string `json:"name"` + Args map[string]interface{} `json:"args"` + } + json.Unmarshal(req.Params, &p) + switch p.Name { + case "fail_tool": + send(response{ID: req.ID, Error: "故意失败：用于验证错误上报"}) + default: + text, _ := p.Args["text"].(string) + send(response{ID: req.ID, Result: map[string]interface{}{"result": text}}) + } + case "event.deliver": + // 通知：不回应答 + default: + if req.ID != 0 { + send(response{ID: req.ID, Error: fmt.Sprintf("未实现 %s", req.Method)}) + } + } + } +} diff --git a/internal/plugin/proc/testdata/hangplugin.go b/internal/plugin/proc/testdata/hangplugin.go new file mode 100644 index 0000000..f98c9d5 --- /dev/null +++ b/internal/plugin/proc/testdata/hangplugin.go @@ -0,0 +1,56 @@ +//go:build ignore + +// hangplugin 收到工具调用后永久阻塞，用于验证： +// 1. 调用方能凭 context 超时返回（不被拖死） +// 2. Kill 能真正回收资源（对比 cgo 超时后 OS 线程永久泄漏，§9.3） +package main + +import ( + "bufio" + "encoding/json" + "os" + "time" +) + +type request struct { + ID uint64 `json:"id,omitempty"` + Method string `json:"method"` + Params json.RawMessage `json:"params,omitempty"` +} + +type response struct { + ID uint64 `json:"id"` + Result interface{} `json:"result,omitempty"` + Error string `json:"error,omitempty"` +} + +func main() { + in := bufio.NewScanner(bufio.NewReader(os.Stdin)) + out := bufio.NewWriter(os.Stdout) + send := func(v interface{}) { + b, _ := json.Marshal(v) + out.Write(b) + out.WriteByte('\n') + out.Flush() + } + + for in.Scan() { + var req request + if err := json.Unmarshal(in.Bytes(), &req); err != nil { + continue + } + switch req.Method { + case "handshake": + send(response{ID: req.ID, Result: map[string]interface{}{ + "protocol": 1, "sdk_version": "test", "plugin_name": "hang", "pid": os.Getpid(), + }}) + case "tool.invoke": + // 永久卡住，永不回应答 + time.Sleep(10 * time.Minute) + default: + if req.ID != 0 { + send(response{ID: req.ID}) + } + } + } +}