package main // sidecar:每个工程对话一份的 pi 运行时外壳。 // // 协议照抄 internal/plugins/clawhubadapter/sidecar.go(本仓已验证的成熟做法): // JSON-RPC over stdin/stdout + pending-id 配对 + readLoop + 优雅 Close。 // 不自创协议 —— 那边踩过的坑(长行截断、pending 表泄漏、退出时唤醒等待者)不必再踩。 // // ## 为什么需要 sidecar // // pi 是 Node 程序。我们**不把 pi 的 RPC 循环搬进 Go**(那是重复实现), // 只用一个薄薄的 Node 脚本把 pi 包成一个「可反复 prompt 的会话」,Go 侧用 // JSON-RPC 跟它说话。这样: // // Go 插件 ──JSON-RPC/stdio──► sidecar(node) ──► pi 会话(内嵌 createAgentSession) // // ## 刀 1 的形态 // // 本地内嵌。远端 ssh 是后续刀次(同一份 sidecar,只换传输层)。 // sidecar 脚本路径:优先随包分发的 runtime/sidecar.mjs,其次插件目录下的 sidecar.mjs。 import ( "bufio" "encoding/json" "fmt" "io" "log" "os" "os/exec" "path/filepath" "strconv" "strings" sdk "github.com/JianFeeeee/homeagentsdk/sdk" "sync" "time" ) type sidecarRequest struct { JSONRPC string `json:"jsonrpc"` ID int `json:"id"` Method string `json:"method"` Params interface{} `json:"params,omitempty"` } type sidecarResponse struct { JSONRPC string `json:"jsonrpc"` ID int `json:"id"` Result *json.RawMessage `json:"result,omitempty"` Error *struct { Code int `json:"code"` Message string `json:"message"` } `json:"error,omitempty"` } type sidecarProcess struct { branchID string workdir string cmd *exec.Cmd stdin *bufio.Writer stdout io.Reader mu sync.Mutex nextID int pending map[int]chan []byte closed bool stopped bool exited chan struct{} waitErr error timeout time.Duration // handshakeTimeout 用于 ping 握手(比普通调用短得多)。 // 为何单列:握手撞上工具调用的 60s 上限会退化成“超时但进程已起”。 handshakeTimeout time.Duration } func newSidecarProcess(branchID, workdir string, cmd *exec.Cmd, stdin *bufio.Writer, stdout io.Reader, timeout time.Duration) *sidecarProcess { sp := &sidecarProcess{ branchID: branchID, workdir: workdir, cmd: cmd, stdin: stdin, stdout: stdout, pending: map[int]chan []byte{}, exited: make(chan struct{}), timeout: timeout, } go sp.readLoop(stdout) go func() { sp.waitErr = cmd.Wait() close(sp.exited) }() return sp } // readLoop 持续读 stdout。 // // ⚠️ 必须持续读:不读会因管道缓冲填满而**堵住 pi**(这是 RPC 类协议最常见的死法)。 func (s *sidecarProcess) readLoop(r io.Reader) { sc := bufio.NewScanner(r) sc.Buffer(make([]byte, 0, 64*1024), 4*1024*1024) for sc.Scan() { line := sc.Bytes() if len(line) == 0 { continue } var resp sidecarResponse if err := json.Unmarshal(line, &resp); err != nil { // 不是协议记录:当诊断信息,不打断 log.Printf("[commondev] sidecar %s 非协议输出: %s", s.branchID, truncate(string(line), 200)) continue } if resp.ID == 0 { // 主动通知(暂未使用,保留扩展位) continue } s.mu.Lock() ch := s.pending[resp.ID] delete(s.pending, resp.ID) s.mu.Unlock() if ch != nil { cp := make([]byte, len(line)) copy(cp, line) ch <- cp // 不 close:避免 close/send 竞态 } } // stdout 结束 ⇒ 进程已亡,唤醒所有等待者 s.mu.Lock() for id, ch := range s.pending { delete(s.pending, id) ch <- nil } s.mu.Unlock() } func (s *sidecarProcess) call(method string, params interface{}) (interface{}, error) { if s.handshakeTimeout > 0 && method == "ping" { return s.callWithTimeout(method, params, s.handshakeTimeout) } return s.callWithTimeout(method, params, s.timeout) } func (s *sidecarProcess) callWithTimeout(method string, params interface{}, timeout time.Duration) (interface{}, error) { raw, err := s.callRawTimeout(method, params, timeout) if err != nil { return nil, err } if raw == nil { return nil, nil } var resp sidecarResponse if err := json.Unmarshal(raw, &resp); err != nil { return nil, fmt.Errorf("sidecar %s 应答解析失败: %w", s.branchID, err) } if resp.Error != nil { return nil, fmt.Errorf("sidecar %s: %s", s.branchID, resp.Error.Message) } if resp.Result == nil { return nil, nil } var out interface{} if err := json.Unmarshal(*resp.Result, &out); err != nil { return nil, err } return out, nil } func (s *sidecarProcess) callRaw(method string, params interface{}) ([]byte, error) { return s.callRawTimeout(method, params, s.timeout) } func (s *sidecarProcess) callRawTimeout(method string, params interface{}, timeout time.Duration) ([]byte, error) { ch := make(chan []byte, 1) s.mu.Lock() if s.closed || s.stopped { s.mu.Unlock() return nil, fmt.Errorf("sidecar %s 已关闭", s.branchID) } s.nextID++ id := s.nextID s.pending[id] = ch req := sidecarRequest{JSONRPC: "2.0", ID: id, Method: method, Params: params} data, err := json.Marshal(req) if err != nil { delete(s.pending, id) s.mu.Unlock() return nil, err } if _, err := s.stdin.Write(data); err != nil { delete(s.pending, id) s.mu.Unlock() return nil, err } if _, err := s.stdin.Write([]byte("\n")); err != nil { delete(s.pending, id) s.mu.Unlock() return nil, err } if err := s.stdin.Flush(); err != nil { delete(s.pending, id) s.mu.Unlock() return nil, err } s.mu.Unlock() select { case raw := <-ch: if raw == nil { return nil, fmt.Errorf("sidecar %s 进程已退出", s.branchID) } return raw, nil case <-time.After(timeout): s.mu.Lock() delete(s.pending, id) s.mu.Unlock() return nil, fmt.Errorf("sidecar %s 调用 %s 超时", s.branchID, method) case <-s.exited: return nil, fmt.Errorf("sidecar %s 进程已退出", s.branchID) } } // Alive 报告进程是否还活着。 func (s *sidecarProcess) Alive() bool { if s == nil { return false } select { case <-s.exited: return false default: return true } } // Close 优雅收尾:先关 stdin 请求有序退出,超时则强杀。 // // ★ 孤儿清理是插件的责任:绝不允许留下 pi 进程。 func (s *sidecarProcess) Close() { if s == nil { return } s.mu.Lock() if s.closed { s.mu.Unlock() return } s.closed = true s.mu.Unlock() // ★ 关 stdin = 请求有序退出(与 clawhubadapter 同纪律)。 // // 这里的 nil 判定不只是防御式编程:Close() 的签名承诺了 nil-safe // (上面 `if s == nil`),但**未完整初始化的对象**(stdin 为 nil // interface)同样会在这里 panic —— 而调用方看不到差别 // (两者都是 *sidecarProcess)。 // // 实测来源:`deliverToPi` / `handleBranchClose` 都可能在 startSidecar // 中途失败时拿到半初始化对象并调 Close;空指针 panic 会盖住真正的错误。 if s.stdin != nil { _ = s.stdin.Flush() } // ★ 若从未真正启动(cmd/exited 为空),没有可等、可杀的东西 —— 直接收尾。 // // 为何必须显式判:Close() 在 `if s == nil` 之后就不再检查内部字段, // 而 nil-safe 的承诺对**半初始化对象**同样适用(调用方只看到 // *sidecarProcess,分不出"从未启动"与"已退出")。 // 实测:startSidecar 中途失败时,handleBranchClose 会拿到这种对象; // 不判则 `s.cmd.Process` 解引用 nil 指针 —— panic 会盖住真正的错误。 if s.cmd == nil || s.exited == nil { s.markStopped() return } select { case <-s.exited: // 已退 case <-time.After(3 * time.Second): if s.cmd.Process != nil { _ = s.cmd.Process.Kill() } select { case <-s.exited: case <-time.After(2 * time.Second): log.Printf("[commondev] sidecar %s 强杀后仍未退出", s.branchID) } } s.markStopped() } // markStopped 置停止标记并唤醒所有挂起的调用(幂等)。 func (s *sidecarProcess) markStopped() { s.mu.Lock() s.stopped = true for id, ch := range s.pending { delete(s.pending, id) ch <- nil } s.mu.Unlock() } // ---------------- 启动 ---------------- // startSidecar 拉起某个工程对话的运行时(**按 agent_cli 选驱动**)。 // // ★ 本函数**不知道 pi 的存在**:它只读 driver.json(或内置 pi 描述), // 按声明起进程、注环境、传参数。换 CLI = 换一份 driver.json,这里一行不改。 // // 契约(sidecar 脚本必须实现,driver 只负责把进程拉起来): // // ping → {ok, session_id, ...} // prompt → {ok, session_id, reply} ← session_id 必须回,Go 侧据此断言会话未漂移 // state → {ok, session_id, messages} // close → {ok} func (p *Plugin) startSidecar(b *Branch, sessionID string) (*sidecarProcess, error) { p.mu.Lock() existing := p.sidecars[b.ID] p.mu.Unlock() if existing != nil && existing.Alive() { return existing, nil // 复用 } cli := strings.TrimSpace(b.AgentCLI) if cli == "" { cli = defaultAgentCLI } drv, dir, err := p.loadDriver(cli) if err != nil { return nil, err } // 脚本查找链(按优先级): // // ① <插件数据>/runtime//