mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-10-15 12:15:16 +00:00
refactor: rename openclaw -> clawhubadapter, add ClawHub search/install, fix agent interrupt
- Rename internal/plugins/openclaw/ -> internal/plugins/clawhubadapter/ - Add RegistryDispatcher with 5 sub-registries (Tool, Provider, Channel, Stage, Cap) - Add clawhubadapter_search tool for ClawHub marketplace search - Add clawhub: prefix support for installing from ClawHub (ZIP/tgz auto-detect) - Add CallProvider RPC and provider/call routing - Fix QQ interrupt: check interceptCh before/after each tool execution
This commit is contained in:
359
internal/plugins/clawhubadapter/sidecar.go
Normal file
359
internal/plugins/clawhubadapter/sidecar.go
Normal file
@ -0,0 +1,359 @@
|
||||
package clawhubadapter
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"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"`
|
||||
}
|
||||
|
||||
type sidecarProcess struct {
|
||||
name string
|
||||
dir string
|
||||
cmd *exec.Cmd
|
||||
stdin *bufio.Writer
|
||||
mu sync.Mutex
|
||||
nextID int
|
||||
closed bool
|
||||
stopped bool
|
||||
|
||||
// 异步 reader
|
||||
pending map[int]chan<- []byte
|
||||
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]chan<- []byte),
|
||||
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)
|
||||
close(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()
|
||||
ch, ok := s.pending[*base.ID]
|
||||
delete(s.pending, *base.ID)
|
||||
s.mu.Unlock()
|
||||
if ok {
|
||||
ch <- []byte(line)
|
||||
close(ch)
|
||||
}
|
||||
} 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 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)
|
||||
}
|
||||
|
||||
func launchProcess(bin, arg, dir, name 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)
|
||||
cmd.Dir = dir
|
||||
cmd.Stderr = os.Stderr
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
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(5 * time.Second):
|
||||
return fmt.Errorf("sidecar %s ping timeout", s.name)
|
||||
}
|
||||
}
|
||||
|
||||
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] = 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, ch := range s.pending {
|
||||
close(ch)
|
||||
delete(s.pending, id)
|
||||
}
|
||||
s.mu.Unlock()
|
||||
|
||||
log.Printf("[clawhubadapter] sidecar %s stopped", s.name)
|
||||
}
|
||||
Reference in New Issue
Block a user