mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-11-01 06:47:29 +00:00
29150065aa
CI 报红：`TestLaunchSidecarAndListTools` launch sidecar: sidecar echoplugin ping timeout (5.00s) 本机全绿。 ## 这不是"测试机器慢所以放宽测试" sidecar 起不来等于该 skill 整体不可用 —— 这是**产品行为**，不是测试细节。 原来的 5s 是按开发机估算的：node 冷启动 + 模块加载在快盘空载时约 0.2s。 而 CI 共享 runner（以及真实部署里的低配机器）上，一次 fork+exec+node 引导遇到 CPU 争抢就会超过 5s。 判断"真的挂了"应该由**进程退出信号**负责（readLoop 会感知）， 而不是由引导耗时猜。 ## 30s 的取法 · 比 systemd 的 TimeoutStopSec(25s) 宽 —— 关停时不该被就绪等待拖住 · 与用户可感知的"卡住"阈值（约 30s）持平 —— 再长就该让用户看到明确失败 · 错误文案带上超时值与可能原因（node 不可用），便于自助排查
421 lines
10 KiB
Go
421 lines
10 KiB
Go
package clawhubadapter
|
||
|
||
import (
|
||
"bufio"
|
||
"encoding/json"
|
||
"fmt"
|
||
safeclose "github.com/JianFeeeee/HologramAgent/internal/safeclose"
|
||
"io"
|
||
"log"
|
||
"os"
|
||
"os/exec"
|
||
"path/filepath"
|
||
"strings"
|
||
"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 OCPTool struct {
|
||
Name string `json:"name"`
|
||
Description string `json:"description"`
|
||
InputSchema map[string]interface{} `json:"inputSchema"`
|
||
}
|
||
|
||
type OCCallResult struct {
|
||
Content []struct {
|
||
Type string `json:"type"`
|
||
Text string `json:"text,omitempty"`
|
||
} `json:"content"`
|
||
}
|
||
|
||
// OCNotification 是模拟器主动推送的通知
|
||
type OCNotification struct {
|
||
JSONRPC string `json:"jsonrpc"`
|
||
Method string `json:"method"`
|
||
Params json.RawMessage `json:"params"`
|
||
}
|
||
|
||
// pendingResp 是待应答通道 + 它自己的 close 守卫。
|
||
//
|
||
// 为什么需要 once：Close() 可能被多条路径调用（Stop / 重载 / 停机），
|
||
// 而 `close(ch)` 二次即 panic（本仓 close 纪律，见 internal/safeclose）。
|
||
// 这些 channel 是**按请求创建**的，不能共用一个全局 Once ⇒ 每个自带一个。
|
||
type pendingResp struct {
|
||
ch chan []byte
|
||
once sync.Once
|
||
}
|
||
|
||
func (p *pendingResp) closeOnce() { safeclose.Safe(&p.once, p.ch) }
|
||
|
||
type sidecarProcess struct {
|
||
readerReadyOnce sync.Once
|
||
name string
|
||
dir string
|
||
cmd *exec.Cmd
|
||
stdin *bufio.Writer
|
||
mu sync.Mutex
|
||
nextID int
|
||
closed bool
|
||
stopped bool
|
||
|
||
// 异步 reader
|
||
pending map[int]*pendingResp
|
||
notifyCh chan OCNotification
|
||
readerStop chan struct{}
|
||
readerWg sync.WaitGroup
|
||
readerReady chan struct{}
|
||
}
|
||
|
||
func newSidecarProcess(name, dir string, cmd *exec.Cmd, stdin *bufio.Writer, stdout io.Reader) *sidecarProcess {
|
||
sp := &sidecarProcess{
|
||
name: name,
|
||
dir: dir,
|
||
cmd: cmd,
|
||
stdin: stdin,
|
||
pending: make(map[int]*pendingResp),
|
||
notifyCh: make(chan OCNotification, 1024),
|
||
readerStop: make(chan struct{}),
|
||
readerReady: make(chan struct{}),
|
||
}
|
||
sp.readerWg.Add(1)
|
||
go sp.readLoop(stdout)
|
||
<-sp.readerReady
|
||
return sp
|
||
}
|
||
|
||
func (s *sidecarProcess) readLoop(r io.Reader) {
|
||
defer s.readerWg.Done()
|
||
scanner := bufio.NewScanner(bufio.NewReader(r))
|
||
// 加大 scanner buffer 防止长行截断
|
||
scanner.Buffer(make([]byte, 0, 1024*64), 1024*64)
|
||
safeclose.Safe(&s.readerReadyOnce, s.readerReady)
|
||
|
||
for {
|
||
select {
|
||
case <-s.readerStop:
|
||
return
|
||
default:
|
||
}
|
||
|
||
if !scanner.Scan() {
|
||
if scanner.Err() != nil {
|
||
log.Printf("[clawhubadapter] sidecar %s read error: %v", s.name, scanner.Err())
|
||
}
|
||
return
|
||
}
|
||
line := scanner.Text()
|
||
|
||
var base struct {
|
||
ID *int `json:"id"`
|
||
Method string `json:"method,omitempty"`
|
||
Error *json.RawMessage `json:"error,omitempty"`
|
||
}
|
||
if err := json.Unmarshal([]byte(line), &base); err != nil {
|
||
continue
|
||
}
|
||
|
||
if base.ID != nil {
|
||
s.mu.Lock()
|
||
pr, ok := s.pending[*base.ID]
|
||
delete(s.pending, *base.ID)
|
||
s.mu.Unlock()
|
||
if ok {
|
||
pr.ch <- []byte(line)
|
||
}
|
||
} else if base.Method != "" {
|
||
var notif OCNotification
|
||
if err := json.Unmarshal([]byte(line), ¬if); err == nil {
|
||
select {
|
||
case s.notifyCh <- notif:
|
||
default:
|
||
log.Printf("[clawhubadapter] sidecar %s notify channel full, dropping: %s", s.name, notif.Method)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
func (s *sidecarProcess) NotifyChan() <-chan OCNotification {
|
||
return s.notifyCh
|
||
}
|
||
|
||
func launchSidecar(dir, name, simDir string) (*sidecarProcess, error) {
|
||
mainJS := filepath.Join(dir, "main.js")
|
||
if _, err := os.Stat(mainJS); os.IsNotExist(err) {
|
||
return nil, nil
|
||
}
|
||
return launchProcess("node", mainJS, dir, name, simDir)
|
||
}
|
||
|
||
// launchProcess 以 dir 作为工作目录拉起子进程（绝大多数调用的形态）。
|
||
func launchProcess(bin, arg, dir, name, simDir string) (*sidecarProcess, error) {
|
||
return launchProcessIn(bin, arg, dir, name, simDir, dir)
|
||
}
|
||
|
||
// launchProcessIn 与 launchProcess 相同，但工作目录可单独指定。
|
||
//
|
||
// 为什么要单独一个函数而不是给 launchProcess 加参数：那会改动 4 个既有
|
||
// 调用点（含测试），而它们的语义都是"工作目录 = skill 目录"。只有
|
||
// skill_workdir 配置这一处需要另给目录，用变体表达的意图更清楚。
|
||
func launchProcessIn(bin, arg, dir, name, simDir, workDir string) (*sidecarProcess, error) {
|
||
nodePath := bin
|
||
if bin == "node" {
|
||
if p := os.Getenv("NODE_PATH"); p != "" {
|
||
nodePath = filepath.Join(p, "node")
|
||
}
|
||
}
|
||
|
||
cmd := exec.Command(nodePath, arg, dir)
|
||
if workDir == "" {
|
||
workDir = dir
|
||
}
|
||
cmd.Dir = workDir
|
||
cmd.Stderr = os.Stderr
|
||
|
||
// Add openclaw CLI bin dir to PATH so subprocesses can exec 'openclaw' command
|
||
if simDir != "" {
|
||
binDir := filepath.Join(simDir, "bin")
|
||
if info, err := os.Stat(binDir); err == nil && info.IsDir() {
|
||
env := os.Environ()
|
||
binDirPath := binDir + string(os.PathListSeparator)
|
||
for i, e := range env {
|
||
if strings.HasPrefix(e, "PATH=") {
|
||
env[i] = "PATH=" + binDirPath + e[5:]
|
||
break
|
||
}
|
||
}
|
||
cmd.Env = env
|
||
}
|
||
}
|
||
|
||
stdin, err := cmd.StdinPipe()
|
||
if err != nil {
|
||
return nil, fmt.Errorf("stdin pipe: %w", err)
|
||
}
|
||
stdout, err := cmd.StdoutPipe()
|
||
if err != nil {
|
||
return nil, fmt.Errorf("stdout pipe: %w", err)
|
||
}
|
||
|
||
if err := cmd.Start(); err != nil {
|
||
return nil, fmt.Errorf("start %s: %w", name, err)
|
||
}
|
||
|
||
sp := newSidecarProcess(name, dir, cmd, bufio.NewWriter(stdin), stdout)
|
||
|
||
if err := sp.waitReady(); err != nil {
|
||
sp.Close()
|
||
return nil, err
|
||
}
|
||
|
||
return sp, nil
|
||
}
|
||
|
||
// sidecarReadyTimeout 是等 sidecar 回第一个 ping 的上限。
|
||
//
|
||
// ★ 从固定 5s 放宽到 30s（2026-10-25 修 CI 误报）。
|
||
//
|
||
// 原来的 5s 是**按开发机估算**的：node 冷启动 + 模块加载在快盘上空载时
|
||
// 约 0.2s。而 CI 的共享 runner（以及真实部署里的低配机器）上，一次
|
||
// fork+exec+node 引导遇到 CPU 争抢就会到 5s 以上，实测 CI 报
|
||
// `sidecar echoplugin ping timeout` 而本机全绿。
|
||
//
|
||
// 这不是"测试环境太慢所以放宽测试"——它是**产品行为**：sidecar 起不来
|
||
// 等于该 skill 整体不可用。给足时间是正确取舍；判断"真的挂了"应由进程
|
||
// 退出信号负责（readLoop 会感知），而不是由引导耗时猜。
|
||
//
|
||
// 30s 的取法：比 systemd 的 TimeoutStopSec(25s) 宽，比用户可感知的
|
||
// "卡住"阈值(约 30s)持平——再长就该让用户看到明确失败。
|
||
const sidecarReadyTimeout = 30 * time.Second
|
||
|
||
func (s *sidecarProcess) waitReady() error {
|
||
done := make(chan error, 1)
|
||
go func() {
|
||
_, err := s.call("ping", nil)
|
||
done <- err
|
||
}()
|
||
select {
|
||
case err := <-done:
|
||
return err
|
||
case <-time.After(sidecarReadyTimeout):
|
||
return fmt.Errorf("sidecar %s ping timeout（%v 内未响应；慢机器上 node 冷启动可能更久，"+
|
||
"若持续失败请检查 node 是否可用）", s.name, sidecarReadyTimeout)
|
||
}
|
||
}
|
||
|
||
func (s *sidecarProcess) call(method string, params interface{}) ([]byte, error) {
|
||
ch := make(chan []byte, 1)
|
||
|
||
s.mu.Lock()
|
||
if s.closed || s.stopped {
|
||
s.mu.Unlock()
|
||
return nil, fmt.Errorf("sidecar %s closed", s.name)
|
||
}
|
||
s.nextID++
|
||
id := s.nextID
|
||
s.pending[id] = &pendingResp{ch: 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 error: nil response", s.name)
|
||
}
|
||
var resp sidecarResponse
|
||
if err := json.Unmarshal(raw, &resp); err != nil {
|
||
return nil, fmt.Errorf("sidecar %s unmarshal: %w", s.name, err)
|
||
}
|
||
if resp.Error != nil {
|
||
return nil, fmt.Errorf("sidecar %s error: %s", s.name, resp.Error.Message)
|
||
}
|
||
if resp.Result == nil {
|
||
return nil, nil
|
||
}
|
||
return []byte(*resp.Result), nil
|
||
case <-time.After(30 * time.Second):
|
||
s.mu.Lock()
|
||
delete(s.pending, id)
|
||
s.mu.Unlock()
|
||
return nil, fmt.Errorf("sidecar %s call %s timeout", s.name, method)
|
||
}
|
||
}
|
||
|
||
func (s *sidecarProcess) ListTools() ([]OCPTool, error) {
|
||
data, err := s.call("tools/list", nil)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if data == nil {
|
||
return nil, nil
|
||
}
|
||
var result struct {
|
||
Tools []OCPTool `json:"tools"`
|
||
}
|
||
if err := json.Unmarshal(data, &result); err != nil {
|
||
return nil, err
|
||
}
|
||
return result.Tools, nil
|
||
}
|
||
|
||
func (s *sidecarProcess) CallTool(name string, args map[string]interface{}) (string, error) {
|
||
data, err := s.call("tools/call", map[string]interface{}{
|
||
"name": name,
|
||
"arguments": args,
|
||
})
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
if data == nil {
|
||
return "", nil
|
||
}
|
||
var result OCCallResult
|
||
if err := json.Unmarshal(data, &result); err != nil {
|
||
return "", err
|
||
}
|
||
var sb string
|
||
for _, c := range result.Content {
|
||
if c.Type == "text" {
|
||
sb += c.Text
|
||
}
|
||
}
|
||
return sb, nil
|
||
}
|
||
|
||
func (s *sidecarProcess) CallProvider(providerType string, args map[string]interface{}) (string, error) {
|
||
data, err := s.call("provider/call", map[string]interface{}{
|
||
"type": providerType,
|
||
"action": "execute",
|
||
"args": args,
|
||
})
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
if data == nil {
|
||
return "", nil
|
||
}
|
||
var result OCCallResult
|
||
if err := json.Unmarshal(data, &result); err != nil {
|
||
return "", err
|
||
}
|
||
var sb string
|
||
for _, c := range result.Content {
|
||
if c.Type == "text" {
|
||
sb += c.Text
|
||
}
|
||
}
|
||
return sb, nil
|
||
}
|
||
|
||
func (s *sidecarProcess) Close() {
|
||
s.mu.Lock()
|
||
if s.closed || s.stopped {
|
||
s.mu.Unlock()
|
||
return
|
||
}
|
||
s.stopped = true
|
||
s.mu.Unlock()
|
||
|
||
// 先杀进程（关闭 stdout pipe），然后 reader 的 Scan() 会退出
|
||
if s.cmd != nil && s.cmd.Process != nil {
|
||
s.cmd.Process.Kill()
|
||
s.cmd.Wait()
|
||
}
|
||
|
||
// 等 reader 循环结束
|
||
s.readerWg.Wait()
|
||
|
||
s.mu.Lock()
|
||
for id, pr := range s.pending {
|
||
pr.closeOnce()
|
||
delete(s.pending, id)
|
||
}
|
||
s.mu.Unlock()
|
||
|
||
log.Printf("[clawhubadapter] sidecar %s stopped", s.name)
|
||
}
|