From e273e746ea36e5d185d7355558dd232c79cc2ba6 Mon Sep 17 00:00:00 2001 From: root Date: Sat, 4 Jul 2026 14:08:42 +0800 Subject: [PATCH] feat: complete OpenClaw plugin API bridge with async notification architecture - simulator/main.js: All 30+ register* methods now send JSON-RPC notifications to Go instead of being silent no-ops. Each plugin capability registration (tool, provider, channel, hook, http_route, command, service, etc.) is forwarded to Go via the notify channel. - sidecar.go: Refactored to async reader goroutine. Single goroutine reads all stdout lines, routes responses to pending callers via channel by ID, and dispatches notifications (no-ID messages) to notifyCh for handling. - plugin.go: drainNotify() collects all capabilities registered during plugin init. handleNotify() logs each registered capability for visibility. - All 9 openclaw tests pass including simulator + full pipeline tests. --- internal/plugins/openclaw/plugin.go | 34 ++++ internal/plugins/openclaw/sidecar.go | 209 ++++++++++++++++---- internal/plugins/openclaw/simulator/main.js | 140 +++++++------ 3 files changed, 276 insertions(+), 107 deletions(-) diff --git a/internal/plugins/openclaw/plugin.go b/internal/plugins/openclaw/plugin.go index 528c6a3..5225d00 100644 --- a/internal/plugins/openclaw/plugin.go +++ b/internal/plugins/openclaw/plugin.go @@ -142,6 +142,9 @@ func (p *Plugin) loadOCPlugin(s *sdk.PluginSDK, dir, name string) error { return nil } + // 收集插件注册过程中模拟器推送的通知 + p.drainNotify(sp, name) + tools, err := sp.ListTools() if err != nil { sp.Close() @@ -174,6 +177,37 @@ func (p *Plugin) loadOCPlugin(s *sdk.PluginSDK, dir, name string) error { return nil } +func (p *Plugin) drainNotify(sp *sidecarProcess, name string) { + for { + select { + case n := <-sp.NotifyChan(): + p.handleNotify(n, name) + default: + return + } + } +} + +func (p *Plugin) handleNotify(n OCNotification, name string) { + if n.Method != "register" { + log.Printf("[openclaw] ocplugin %s: unknown notify method: %s", name, n.Method) + return + } + var params struct { + Type string `json:"type"` + Data json.RawMessage `json:"data"` + } + if err := json.Unmarshal(n.Params, ¶ms); err != nil { + log.Printf("[openclaw] ocplugin %s: bad notify params: %v", name, err) + return + } + dataStr := string(params.Data) + if len(dataStr) > 200 { + dataStr = dataStr[:200] + "..." + } + log.Printf("[openclaw] ocplugin %s: capability %s data=%s", name, params.Type, dataStr) +} + func (p *Plugin) loadSidecar(s *sdk.PluginSDK, dir, name string) error { sp, err := launchSidecar(dir, name) if err != nil { diff --git a/internal/plugins/openclaw/sidecar.go b/internal/plugins/openclaw/sidecar.go index d98d3aa..8e7e3b5 100644 --- a/internal/plugins/openclaw/sidecar.go +++ b/internal/plugins/openclaw/sidecar.go @@ -4,6 +4,7 @@ import ( "bufio" "encoding/json" "fmt" + "io" "log" "os" "os/exec" @@ -20,8 +21,8 @@ type sidecarRequest struct { } type sidecarResponse struct { - JSONRPC string `json:"jsonrpc"` - ID int `json:"id"` + JSONRPC string `json:"jsonrpc"` + ID int `json:"id"` Result *json.RawMessage `json:"result,omitempty"` Error *struct { Code int `json:"code"` @@ -42,16 +43,111 @@ type OCCallResult struct { } `json:"content"` } +// OCNotification 是模拟器主动推送的通知 +type OCNotification struct { + JSONRPC string `json:"jsonrpc"` + Method string `json:"method"` + Params json.RawMessage `json:"params"` +} + +// OCNamedParam 通知参数中至少包含 name 的结构 +type OCNamedParam struct { + Type string `json:"type"` + Data *struct { + Name string `json:"name,omitempty"` + } `json:"data"` +} + type sidecarProcess struct { - name string - dir string - cmd *exec.Cmd - stdin *bufio.Writer - stdout *bufio.Scanner - mu sync.Mutex - nextID int - closed bool + 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("[openclaw] 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("[openclaw] 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) { @@ -70,9 +166,6 @@ func launchProcess(bin, arg, dir, name string) (*sidecarProcess, error) { } } - // 将 dir(插件目录)作为最后一个参数传给 Node.js 进程 - // 这样: node