From 85b014a45c8267f68fcca5d9dd91f9a9182d283a Mon Sep 17 00:00:00 2001 From: zhouqingyi01 Date: Mon, 6 Jul 2026 19:25:14 +0800 Subject: [PATCH 1/3] feat: add moltybot bridge agent --- agent/moltybot/moltybot.go | 109 ++++++++++++ agent/moltybot/moltybot_test.go | 66 +++++++ agent/moltybot/session.go | 224 ++++++++++++++++++++++++ cmd/cc-connect/plugin_agent_moltybot.go | 5 + 4 files changed, 404 insertions(+) create mode 100644 agent/moltybot/moltybot.go create mode 100644 agent/moltybot/moltybot_test.go create mode 100644 agent/moltybot/session.go create mode 100644 cmd/cc-connect/plugin_agent_moltybot.go diff --git a/agent/moltybot/moltybot.go b/agent/moltybot/moltybot.go new file mode 100644 index 000000000..1569823aa --- /dev/null +++ b/agent/moltybot/moltybot.go @@ -0,0 +1,109 @@ +// Package moltybot forwards cc-connect platform messages to a local MoltyBot +// bridge. It does not spawn or drive a coding-agent process. +package moltybot + +import ( + "context" + "fmt" + "log/slog" + "net/http" + "net/url" + "strings" + "time" + + "github.com/chenhg5/cc-connect/core" +) + +const ( + agentName = "moltybot" + defaultSessionMode = "per_remote_user" + sessionModePerRemoteUser = "per_remote_user" + sessionModePassthrough = "passthrough" + defaultBridgeRequestTimeout = 2 * time.Minute + bridgeMessagesEndpointSuffix = "/v1/messages" +) + +func init() { + core.RegisterAgent(agentName, New) +} + +// Agent forwards turns to MoltyBot's local HTTP bridge. +type Agent struct { + baseURL string + token string + sessionMode string + client *http.Client +} + +// New creates a MoltyBot bridge agent. +// +// Required option: +// - base_url: local MoltyBot bridge base URL, for example http://127.0.0.1:48999 +// +// Optional options: +// - token: bearer token sent to the bridge +// - session_mode: per_remote_user (default) or passthrough +func New(opts map[string]any) (core.Agent, error) { + baseURL, _ := opts["base_url"].(string) + baseURL = strings.TrimRight(strings.TrimSpace(baseURL), "/") + if baseURL == "" { + return nil, fmt.Errorf("moltybot: base_url is required") + } + parsed, err := url.Parse(baseURL) + if err != nil { + return nil, fmt.Errorf("moltybot: invalid base_url %q: %w", baseURL, err) + } + if parsed.Scheme == "" || parsed.Host == "" { + return nil, fmt.Errorf("moltybot: invalid base_url %q: missing scheme or host", baseURL) + } + + token, _ := opts["token"].(string) + sessionMode, err := parseSessionMode(opts["session_mode"]) + if err != nil { + return nil, err + } + + slog.Info( + "moltybot: agent created", + "base_url", baseURL, + "session_mode", sessionMode, + "token_configured", strings.TrimSpace(token) != "", + ) + return &Agent{ + baseURL: baseURL, + token: strings.TrimSpace(token), + sessionMode: sessionMode, + client: &http.Client{ + Timeout: defaultBridgeRequestTimeout, + }, + }, nil +} + +func parseSessionMode(raw any) (string, error) { + value, _ := raw.(string) + value = strings.ToLower(strings.TrimSpace(value)) + if value == "" { + return defaultSessionMode, nil + } + switch value { + case sessionModePerRemoteUser, sessionModePassthrough: + return value, nil + default: + return "", fmt.Errorf("moltybot: unsupported session_mode %q", value) + } +} + +func (a *Agent) Name() string { return agentName } + +func (a *Agent) StartSession(ctx context.Context, sessionID string) (core.AgentSession, error) { + slog.Info("moltybot: starting session", "session_id", sessionID, "session_mode", a.sessionMode) + return newSession(ctx, a.client, a.baseURL, a.token, a.sessionMode, sessionID), nil +} + +func (a *Agent) ListSessions(_ context.Context) ([]core.AgentSessionInfo, error) { + return nil, nil +} + +func (a *Agent) Stop() error { return nil } + +var _ core.Agent = (*Agent)(nil) diff --git a/agent/moltybot/moltybot_test.go b/agent/moltybot/moltybot_test.go new file mode 100644 index 000000000..2ebcd0271 --- /dev/null +++ b/agent/moltybot/moltybot_test.go @@ -0,0 +1,66 @@ +package moltybot + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/chenhg5/cc-connect/core" +) + +func TestNewRequiresBaseURL(t *testing.T) { + _, err := New(map[string]any{"token": "secret"}) + if err == nil { + t.Fatal("New returned nil error, want missing base_url error") + } +} + +func TestSessionSendPostsToBridge(t *testing.T) { + var authHeader string + var request map[string]any + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + authHeader = r.Header.Get("Authorization") + if r.URL.Path != "/v1/messages" { + t.Fatalf("path = %s, want /v1/messages", r.URL.Path) + } + if err := json.NewDecoder(r.Body).Decode(&request); err != nil { + t.Fatalf("decode request: %v", err) + } + _ = json.NewEncoder(w).Encode(map[string]any{ + "ok": true, + "sessionKey": "remote:weixin:u1", + "replyText": "收到", + }) + })) + defer server.Close() + + agent, err := New(map[string]any{ + "base_url": server.URL, + "token": "secret", + "session_mode": "per_remote_user", + }) + if err != nil { + t.Fatalf("New returned error: %v", err) + } + + session, err := agent.StartSession(context.Background(), "weixin:chat:u1") + if err != nil { + t.Fatalf("StartSession returned error: %v", err) + } + if err := session.Send("你好", nil, nil); err != nil { + t.Fatalf("Send returned error: %v", err) + } + + ev := <-session.Events() + if ev.Type != core.EventResult || ev.Content != "收到" { + t.Fatalf("event = %#v, want result 收到", ev) + } + if authHeader != "Bearer secret" { + t.Fatalf("Authorization = %q", authHeader) + } + if request["text"] != "你好" { + t.Fatalf("text = %#v", request["text"]) + } +} diff --git a/agent/moltybot/session.go b/agent/moltybot/session.go new file mode 100644 index 000000000..ad89141ea --- /dev/null +++ b/agent/moltybot/session.go @@ -0,0 +1,224 @@ +package moltybot + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "log/slog" + "net/http" + "strings" + "sync" + "sync/atomic" + + "github.com/chenhg5/cc-connect/core" +) + +type session struct { + client *http.Client + baseURL string + token string + sessionMode string + + ctx context.Context + cancel context.CancelFunc + events chan core.Event + alive atomic.Bool + + mu sync.Mutex + sessionID string + closeOnce sync.Once +} + +type bridgeMessageRequest struct { + SessionKey string `json:"sessionKey,omitempty"` + SessionID string `json:"sessionId,omitempty"` + Text string `json:"text"` + Images []bridgeAttachment `json:"images,omitempty"` + Files []bridgeAttachment `json:"files,omitempty"` +} + +type bridgeAttachment struct { + MimeType string `json:"mimeType,omitempty"` + FileName string `json:"fileName,omitempty"` + Data []byte `json:"data,omitempty"` +} + +type bridgeMessageResponse struct { + OK bool `json:"ok"` + SessionKey string `json:"sessionKey,omitempty"` + ReplyText string `json:"replyText,omitempty"` + Error string `json:"error,omitempty"` +} + +func newSession(ctx context.Context, client *http.Client, baseURL, token, sessionMode, sessionID string) *session { + sessionCtx, cancel := context.WithCancel(ctx) + bridgeSessionID := bridgeSessionKey(sessionMode, sessionID) + s := &session{ + client: client, + baseURL: baseURL, + token: token, + sessionMode: sessionMode, + ctx: sessionCtx, + cancel: cancel, + events: make(chan core.Event, 128), + sessionID: bridgeSessionID, + } + s.alive.Store(true) + return s +} + +func (s *session) Send(prompt string, images []core.ImageAttachment, files []core.FileAttachment) error { + if !s.alive.Load() { + return fmt.Errorf("moltybot: session is closed") + } + + sessionID := s.CurrentSessionID() + body := bridgeMessageRequest{ + SessionKey: sessionID, + SessionID: sessionID, + Text: prompt, + Images: convertImageAttachments(images), + Files: convertFileAttachments(files), + } + + var payload bytes.Buffer + if err := json.NewEncoder(&payload).Encode(body); err != nil { + return fmt.Errorf("moltybot: encode bridge request: %w", err) + } + + req, err := http.NewRequestWithContext(s.ctx, http.MethodPost, s.baseURL+bridgeMessagesEndpointSuffix, &payload) + if err != nil { + return fmt.Errorf("moltybot: create bridge request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + if s.token != "" { + req.Header.Set("Authorization", "Bearer "+s.token) + } + + resp, err := s.client.Do(req) + if err != nil { + return fmt.Errorf("moltybot: post bridge message: %s", core.RedactToken(err.Error(), s.token)) + } + defer resp.Body.Close() + + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + raw, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) + detail := strings.TrimSpace(core.RedactToken(string(raw), s.token)) + if detail == "" { + detail = resp.Status + } + return fmt.Errorf("moltybot: bridge returned HTTP %d: %s", resp.StatusCode, detail) + } + + var result bridgeMessageResponse + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return fmt.Errorf("moltybot: decode bridge response: %s", core.RedactToken(err.Error(), s.token)) + } + if !result.OK { + detail := strings.TrimSpace(core.RedactToken(result.Error, s.token)) + if detail == "" { + detail = "unknown bridge error" + } + return fmt.Errorf("moltybot: bridge rejected message: %s", detail) + } + + if result.SessionKey != "" { + s.mu.Lock() + s.sessionID = result.SessionKey + s.mu.Unlock() + } + + s.emit(core.Event{ + Type: core.EventResult, + Content: result.ReplyText, + SessionID: s.CurrentSessionID(), + Done: true, + }) + slog.Debug("moltybot: bridge message completed", "session_id", s.CurrentSessionID()) + return nil +} + +func (s *session) RespondPermission(_ string, _ core.PermissionResult) error { + return fmt.Errorf("moltybot: permission requests are not supported: %w", core.ErrNotSupported) +} + +func (s *session) Events() <-chan core.Event { return s.events } + +func (s *session) CurrentSessionID() string { + s.mu.Lock() + defer s.mu.Unlock() + return s.sessionID +} + +func (s *session) Alive() bool { return s.alive.Load() } + +func (s *session) Close() error { + s.closeOnce.Do(func() { + s.alive.Store(false) + s.cancel() + close(s.events) + }) + return nil +} + +func (s *session) emit(ev core.Event) { + defer func() { _ = recover() }() + select { + case s.events <- ev: + case <-s.ctx.Done(): + } +} + +func bridgeSessionKey(sessionMode, sessionID string) string { + sessionID = strings.TrimSpace(sessionID) + if sessionID == "" { + return "" + } + if sessionMode != sessionModePerRemoteUser || strings.HasPrefix(sessionID, "remote:") { + return sessionID + } + parts := strings.Split(sessionID, ":") + if len(parts) < 2 { + return "remote:" + sessionID + } + platform := strings.TrimSpace(parts[0]) + user := strings.TrimSpace(parts[len(parts)-1]) + if platform == "" || user == "" { + return "remote:" + sessionID + } + return "remote:" + platform + ":" + user +} + +func convertImageAttachments(images []core.ImageAttachment) []bridgeAttachment { + if len(images) == 0 { + return nil + } + out := make([]bridgeAttachment, 0, len(images)) + for _, image := range images { + out = append(out, bridgeAttachment{ + MimeType: image.MimeType, + FileName: image.FileName, + Data: image.Data, + }) + } + return out +} + +func convertFileAttachments(files []core.FileAttachment) []bridgeAttachment { + if len(files) == 0 { + return nil + } + out := make([]bridgeAttachment, 0, len(files)) + for _, file := range files { + out = append(out, bridgeAttachment{ + MimeType: file.MimeType, + FileName: file.FileName, + Data: file.Data, + }) + } + return out +} + +var _ core.AgentSession = (*session)(nil) diff --git a/cmd/cc-connect/plugin_agent_moltybot.go b/cmd/cc-connect/plugin_agent_moltybot.go new file mode 100644 index 000000000..4ca253b3d --- /dev/null +++ b/cmd/cc-connect/plugin_agent_moltybot.go @@ -0,0 +1,5 @@ +//go:build !no_moltybot + +package main + +import _ "github.com/chenhg5/cc-connect/agent/moltybot" From 742125eb54f6672e1b356bc4296fbf8a44e65024 Mon Sep 17 00:00:00 2001 From: zhouqingyi01 Date: Mon, 6 Jul 2026 19:55:13 +0800 Subject: [PATCH 2/3] fix: align moltybot bridge payload --- agent/moltybot/moltybot.go | 8 ++- agent/moltybot/moltybot_test.go | 92 +++++++++++++++++++++++++++- agent/moltybot/session.go | 103 ++++++++++++++++++++++---------- 3 files changed, 167 insertions(+), 36 deletions(-) diff --git a/agent/moltybot/moltybot.go b/agent/moltybot/moltybot.go index 1569823aa..6afa090aa 100644 --- a/agent/moltybot/moltybot.go +++ b/agent/moltybot/moltybot.go @@ -39,9 +39,9 @@ type Agent struct { // // Required option: // - base_url: local MoltyBot bridge base URL, for example http://127.0.0.1:48999 +// - token: bearer token sent to the bridge // // Optional options: -// - token: bearer token sent to the bridge // - session_mode: per_remote_user (default) or passthrough func New(opts map[string]any) (core.Agent, error) { baseURL, _ := opts["base_url"].(string) @@ -58,6 +58,10 @@ func New(opts map[string]any) (core.Agent, error) { } token, _ := opts["token"].(string) + token = strings.TrimSpace(token) + if token == "" { + return nil, fmt.Errorf("moltybot: token is required") + } sessionMode, err := parseSessionMode(opts["session_mode"]) if err != nil { return nil, err @@ -71,7 +75,7 @@ func New(opts map[string]any) (core.Agent, error) { ) return &Agent{ baseURL: baseURL, - token: strings.TrimSpace(token), + token: token, sessionMode: sessionMode, client: &http.Client{ Timeout: defaultBridgeRequestTimeout, diff --git a/agent/moltybot/moltybot_test.go b/agent/moltybot/moltybot_test.go index 2ebcd0271..66936b264 100644 --- a/agent/moltybot/moltybot_test.go +++ b/agent/moltybot/moltybot_test.go @@ -5,7 +5,9 @@ import ( "encoding/json" "net/http" "net/http/httptest" + "strings" "testing" + "time" "github.com/chenhg5/cc-connect/core" ) @@ -17,6 +19,13 @@ func TestNewRequiresBaseURL(t *testing.T) { } } +func TestNewRequiresToken(t *testing.T) { + _, err := New(map[string]any{"base_url": "http://127.0.0.1:12345", "token": " "}) + if err == nil { + t.Fatal("New returned nil error, want missing token error") + } +} + func TestSessionSendPostsToBridge(t *testing.T) { var authHeader string var request map[string]any @@ -49,7 +58,11 @@ func TestSessionSendPostsToBridge(t *testing.T) { if err != nil { t.Fatalf("StartSession returned error: %v", err) } - if err := session.Send("你好", nil, nil); err != nil { + if err := session.Send( + "你好", + []core.ImageAttachment{{MimeType: "image/png", FileName: "pic.png", Data: []byte{1, 2, 3}}}, + []core.FileAttachment{{MimeType: "text/plain", FileName: "note.txt", Data: []byte("hi")}}, + ); err != nil { t.Fatalf("Send returned error: %v", err) } @@ -63,4 +76,81 @@ func TestSessionSendPostsToBridge(t *testing.T) { if request["text"] != "你好" { t.Fatalf("text = %#v", request["text"]) } + if _, ok := request["sessionKey"]; ok { + t.Fatalf("unexpected top-level sessionKey in request: %#v", request["sessionKey"]) + } + if _, ok := request["sessionId"]; ok { + t.Fatalf("unexpected top-level sessionId in request: %#v", request["sessionId"]) + } + if _, ok := request["images"]; ok { + t.Fatalf("unexpected top-level images in request: %#v", request["images"]) + } + if _, ok := request["files"]; ok { + t.Fatalf("unexpected top-level files in request: %#v", request["files"]) + } + source, ok := request["source"].(map[string]any) + if !ok { + t.Fatalf("source = %#v, want object", request["source"]) + } + if source["platform"] != "weixin" { + t.Fatalf("source.platform = %#v, want weixin", source["platform"]) + } + if source["platformUserId"] != "u1" { + t.Fatalf("source.platformUserId = %#v, want u1", source["platformUserId"]) + } + attachments, ok := request["attachments"].([]any) + if !ok || len(attachments) != 2 { + t.Fatalf("attachments = %#v, want 2 attachments", request["attachments"]) + } + image := attachments[0].(map[string]any) + if image["kind"] != "image" || image["name"] != "pic.png" || image["mimeType"] != "image/png" || image["dataBase64"] != "AQID" { + t.Fatalf("image attachment = %#v", image) + } + file := attachments[1].(map[string]any) + if file["kind"] != "file" || file["name"] != "note.txt" || file["mimeType"] != "text/plain" || file["dataBase64"] != "aGk=" { + t.Fatalf("file attachment = %#v", file) + } +} + +func TestSessionSendEmitsEventErrorOnBridgeError(t *testing.T) { + const token = "secret-token" + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _ = json.NewEncoder(w).Encode(map[string]any{ + "ok": false, + "error": "boom " + token, + }) + })) + defer server.Close() + + agent, err := New(map[string]any{ + "base_url": server.URL, + "token": token, + }) + if err != nil { + t.Fatalf("New returned error: %v", err) + } + + session, err := agent.StartSession(context.Background(), "weixin:chat:u1") + if err != nil { + t.Fatalf("StartSession returned error: %v", err) + } + err = session.Send("你好", nil, nil) + if err != nil && strings.Contains(err.Error(), token) { + t.Fatalf("Send error leaked token: %v", err) + } + + select { + case ev := <-session.Events(): + if ev.Type != core.EventError { + t.Fatalf("event type = %s, want %s", ev.Type, core.EventError) + } + if ev.Error == nil { + t.Fatal("event error is nil") + } + if strings.Contains(ev.Error.Error(), token) || strings.Contains(ev.Content, token) { + t.Fatalf("EventError leaked token: %#v", ev) + } + case <-time.After(time.Second): + t.Fatal("timeout waiting for EventError") + } } diff --git a/agent/moltybot/session.go b/agent/moltybot/session.go index ad89141ea..133e32547 100644 --- a/agent/moltybot/session.go +++ b/agent/moltybot/session.go @@ -3,6 +3,7 @@ package moltybot import ( "bytes" "context" + "encoding/base64" "encoding/json" "fmt" "io" @@ -28,21 +29,27 @@ type session struct { mu sync.Mutex sessionID string + source bridgeSource closeOnce sync.Once } type bridgeMessageRequest struct { - SessionKey string `json:"sessionKey,omitempty"` - SessionID string `json:"sessionId,omitempty"` - Text string `json:"text"` - Images []bridgeAttachment `json:"images,omitempty"` - Files []bridgeAttachment `json:"files,omitempty"` + Source bridgeSource `json:"source"` + Text string `json:"text"` + Attachments []bridgeAttachment `json:"attachments,omitempty"` +} + +type bridgeSource struct { + Platform string `json:"platform,omitempty"` + PlatformUserID string `json:"platformUserId,omitempty"` + MessageID string `json:"messageId,omitempty"` } type bridgeAttachment struct { - MimeType string `json:"mimeType,omitempty"` - FileName string `json:"fileName,omitempty"` - Data []byte `json:"data,omitempty"` + Kind string `json:"kind"` + Name string `json:"name,omitempty"` + MimeType string `json:"mimeType,omitempty"` + DataBase64 string `json:"dataBase64,omitempty"` } type bridgeMessageResponse struct { @@ -64,6 +71,7 @@ func newSession(ctx context.Context, client *http.Client, baseURL, token, sessio cancel: cancel, events: make(chan core.Event, 128), sessionID: bridgeSessionID, + source: bridgeSourceFromSessionID(sessionID), } s.alive.Store(true) return s @@ -74,13 +82,10 @@ func (s *session) Send(prompt string, images []core.ImageAttachment, files []cor return fmt.Errorf("moltybot: session is closed") } - sessionID := s.CurrentSessionID() body := bridgeMessageRequest{ - SessionKey: sessionID, - SessionID: sessionID, - Text: prompt, - Images: convertImageAttachments(images), - Files: convertFileAttachments(files), + Source: s.Source(), + Text: prompt, + Attachments: convertAttachments(images, files), } var payload bytes.Buffer @@ -109,7 +114,7 @@ func (s *session) Send(prompt string, images []core.ImageAttachment, files []cor if detail == "" { detail = resp.Status } - return fmt.Errorf("moltybot: bridge returned HTTP %d: %s", resp.StatusCode, detail) + return s.emitError(fmt.Errorf("moltybot: bridge returned HTTP %d: %s", resp.StatusCode, detail)) } var result bridgeMessageResponse @@ -121,7 +126,7 @@ func (s *session) Send(prompt string, images []core.ImageAttachment, files []cor if detail == "" { detail = "unknown bridge error" } - return fmt.Errorf("moltybot: bridge rejected message: %s", detail) + return s.emitError(fmt.Errorf("moltybot: bridge rejected message: %s", detail)) } if result.SessionKey != "" { @@ -152,6 +157,12 @@ func (s *session) CurrentSessionID() string { return s.sessionID } +func (s *session) Source() bridgeSource { + s.mu.Lock() + defer s.mu.Unlock() + return s.source +} + func (s *session) Alive() bool { return s.alive.Load() } func (s *session) Close() error { @@ -171,6 +182,17 @@ func (s *session) emit(ev core.Event) { } } +func (s *session) emitError(err error) error { + s.emit(core.Event{ + Type: core.EventError, + Content: err.Error(), + SessionID: s.CurrentSessionID(), + Done: true, + Error: err, + }) + return err +} + func bridgeSessionKey(sessionMode, sessionID string) string { sessionID = strings.TrimSpace(sessionID) if sessionID == "" { @@ -191,31 +213,46 @@ func bridgeSessionKey(sessionMode, sessionID string) string { return "remote:" + platform + ":" + user } -func convertImageAttachments(images []core.ImageAttachment) []bridgeAttachment { - if len(images) == 0 { - return nil +func bridgeSourceFromSessionID(sessionID string) bridgeSource { + sessionID = strings.TrimSpace(sessionID) + if sessionID == "" { + return bridgeSource{} } - out := make([]bridgeAttachment, 0, len(images)) - for _, image := range images { - out = append(out, bridgeAttachment{ - MimeType: image.MimeType, - FileName: image.FileName, - Data: image.Data, - }) + if strings.HasPrefix(sessionID, "remote:") { + parts := strings.SplitN(sessionID, ":", 3) + if len(parts) == 3 { + return bridgeSource{Platform: parts[1], PlatformUserID: parts[2]} + } + } + parts := strings.Split(sessionID, ":") + if len(parts) == 1 { + return bridgeSource{Platform: parts[0], PlatformUserID: parts[0]} + } + return bridgeSource{ + Platform: parts[0], + PlatformUserID: parts[len(parts)-1], } - return out } -func convertFileAttachments(files []core.FileAttachment) []bridgeAttachment { - if len(files) == 0 { +func convertAttachments(images []core.ImageAttachment, files []core.FileAttachment) []bridgeAttachment { + if len(images) == 0 && len(files) == 0 { return nil } - out := make([]bridgeAttachment, 0, len(files)) + out := make([]bridgeAttachment, 0, len(images)+len(files)) + for _, image := range images { + out = append(out, bridgeAttachment{ + Kind: "image", + Name: image.FileName, + MimeType: image.MimeType, + DataBase64: base64.StdEncoding.EncodeToString(image.Data), + }) + } for _, file := range files { out = append(out, bridgeAttachment{ - MimeType: file.MimeType, - FileName: file.FileName, - Data: file.Data, + Kind: "file", + Name: file.FileName, + MimeType: file.MimeType, + DataBase64: base64.StdEncoding.EncodeToString(file.Data), }) } return out From bbbbbfc8876cdfbe6c585ae1e52200408865c42a Mon Sep 17 00:00:00 2001 From: zhouqingyi01 Date: Wed, 8 Jul 2026 14:33:16 +0800 Subject: [PATCH 3/3] fix: deliver moltybot bridge attachments --- agent/moltybot/moltybot_test.go | 92 +++++++++++++++++++++++++++++++++ agent/moltybot/session.go | 54 +++++++++++++++++-- core/engine.go | 60 ++++++++++++++++++++- core/engine_test.go | 47 ++++++++++++++--- core/message.go | 30 ++++++----- 5 files changed, 257 insertions(+), 26 deletions(-) diff --git a/agent/moltybot/moltybot_test.go b/agent/moltybot/moltybot_test.go index 66936b264..158bb8fb0 100644 --- a/agent/moltybot/moltybot_test.go +++ b/agent/moltybot/moltybot_test.go @@ -112,6 +112,98 @@ func TestSessionSendPostsToBridge(t *testing.T) { } } +func TestSessionSendEmitsBridgeResponseAttachments(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _ = json.NewEncoder(w).Encode(map[string]any{ + "ok": true, + "sessionKey": "remote:weixin:u1", + "replyText": "已生成。", + "attachments": []map[string]any{ + {"kind": "image", "name": "generated.png", "mimeType": "image/png", "dataBase64": "AQID"}, + {"kind": "file", "name": "report.txt", "mimeType": "text/plain", "dataBase64": "aGk="}, + }, + }) + })) + defer server.Close() + + agent, err := New(map[string]any{ + "base_url": server.URL, + "token": "secret", + }) + if err != nil { + t.Fatalf("New returned error: %v", err) + } + session, err := agent.StartSession(context.Background(), "weixin:chat:u1") + if err != nil { + t.Fatalf("StartSession returned error: %v", err) + } + if err := session.Send("画图", nil, nil); err != nil { + t.Fatalf("Send returned error: %v", err) + } + + ev := <-session.Events() + if ev.Type != core.EventResult || ev.Content != "已生成。" { + t.Fatalf("event = %#v, want result 已生成。", ev) + } + if len(ev.Images) != 1 || ev.Images[0].FileName != "generated.png" || string(ev.Images[0].Data) != "\x01\x02\x03" { + t.Fatalf("images = %#v", ev.Images) + } + if len(ev.Files) != 1 || ev.Files[0].FileName != "report.txt" || string(ev.Files[0].Data) != "hi" { + t.Fatalf("files = %#v", ev.Files) + } +} + +func TestSessionSendUsesInjectedCCSessionKeyWhenStartingFresh(t *testing.T) { + var request map[string]any + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if err := json.NewDecoder(r.Body).Decode(&request); err != nil { + t.Fatalf("decode request: %v", err) + } + _ = json.NewEncoder(w).Encode(map[string]any{ + "ok": true, + "sessionKey": "remote:weixin:u1", + "replyText": "收到", + }) + })) + defer server.Close() + + agent, err := New(map[string]any{ + "base_url": server.URL, + "token": "secret", + "session_mode": "per_remote_user", + }) + if err != nil { + t.Fatalf("New returned error: %v", err) + } + inj, ok := agent.(core.SessionEnvInjector) + if !ok { + t.Fatal("moltybot agent does not implement core.SessionEnvInjector") + } + inj.SetSessionEnv([]string{"CC_PROJECT=moltybot", "CC_SESSION_KEY=weixin:dm:u1"}) + + session, err := agent.StartSession(context.Background(), "") + if err != nil { + t.Fatalf("StartSession returned error: %v", err) + } + if got := session.CurrentSessionID(); got != "remote:weixin:u1" { + t.Fatalf("CurrentSessionID = %q, want remote:weixin:u1", got) + } + if err := session.Send("你好", nil, nil); err != nil { + t.Fatalf("Send returned error: %v", err) + } + + source, ok := request["source"].(map[string]any) + if !ok { + t.Fatalf("source = %#v, want object", request["source"]) + } + if source["platform"] != "weixin" { + t.Fatalf("source.platform = %#v, want weixin", source["platform"]) + } + if source["platformUserId"] != "u1" { + t.Fatalf("source.platformUserId = %#v, want u1", source["platformUserId"]) + } +} + func TestSessionSendEmitsEventErrorOnBridgeError(t *testing.T) { const token = "secret-token" server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { diff --git a/agent/moltybot/session.go b/agent/moltybot/session.go index 133e32547..8d72dae28 100644 --- a/agent/moltybot/session.go +++ b/agent/moltybot/session.go @@ -53,10 +53,11 @@ type bridgeAttachment struct { } type bridgeMessageResponse struct { - OK bool `json:"ok"` - SessionKey string `json:"sessionKey,omitempty"` - ReplyText string `json:"replyText,omitempty"` - Error string `json:"error,omitempty"` + OK bool `json:"ok"` + SessionKey string `json:"sessionKey,omitempty"` + ReplyText string `json:"replyText,omitempty"` + Attachments []bridgeAttachment `json:"attachments,omitempty"` + Error string `json:"error,omitempty"` } func newSession(ctx context.Context, client *http.Client, baseURL, token, sessionMode, sessionID string) *session { @@ -134,10 +135,16 @@ func (s *session) Send(prompt string, images []core.ImageAttachment, files []cor s.sessionID = result.SessionKey s.mu.Unlock() } + resultImages, resultFiles, err := convertResponseAttachments(result.Attachments) + if err != nil { + return s.emitError(fmt.Errorf("moltybot: decode bridge response attachments: %w", err)) + } s.emit(core.Event{ Type: core.EventResult, Content: result.ReplyText, + Images: resultImages, + Files: resultFiles, SessionID: s.CurrentSessionID(), Done: true, }) @@ -258,4 +265,43 @@ func convertAttachments(images []core.ImageAttachment, files []core.FileAttachme return out } +func convertResponseAttachments(attachments []bridgeAttachment) ([]core.ImageAttachment, []core.FileAttachment, error) { + if len(attachments) == 0 { + return nil, nil, nil + } + images := make([]core.ImageAttachment, 0) + files := make([]core.FileAttachment, 0) + for _, attachment := range attachments { + data, err := base64.StdEncoding.DecodeString(strings.TrimSpace(attachment.DataBase64)) + if err != nil { + return nil, nil, fmt.Errorf("attachment %q: invalid base64: %w", attachment.Name, err) + } + if len(data) == 0 { + continue + } + kind := strings.ToLower(strings.TrimSpace(attachment.Kind)) + mimeType := strings.TrimSpace(attachment.MimeType) + if kind == "" && strings.HasPrefix(strings.ToLower(mimeType), "image/") { + kind = "image" + } + switch kind { + case "image": + images = append(images, core.ImageAttachment{ + MimeType: mimeType, + FileName: attachment.Name, + Data: data, + }) + case "file": + files = append(files, core.FileAttachment{ + MimeType: mimeType, + FileName: attachment.Name, + Data: data, + }) + default: + return nil, nil, fmt.Errorf("attachment %q: unsupported kind %q", attachment.Name, attachment.Kind) + } + } + return images, files, nil +} + var _ core.AgentSession = (*session)(nil) diff --git a/core/engine.go b/core/engine.go index 635c76ccf..bae442585 100644 --- a/core/engine.go +++ b/core/engine.go @@ -4428,6 +4428,12 @@ func (e *Engine) runUnsolicitedReader(ctx context.Context, cancel context.Cancel e.send(p, replyCtx, chunk) } } + if len(event.Images) > 0 || len(event.Files) > 0 { + if err := e.sendEventResultAttachments(p, replyCtx, event.Images, event.Files); err != nil { + slog.Error("unsolicited: failed to send EventResult attachments", "platform", p.Name(), "error", err) + e.send(p, replyCtx, fmt.Sprintf(e.i18n.T(MsgError), err)) + } + } // Safety note: concurrent writes to session.History by the // unsolicited reader and a foreground turn cannot overlap. @@ -5285,6 +5291,7 @@ func (e *Engine) processInteractiveEvents(state *interactiveState, session *Sess state.eventsNeedResync = false state.mu.Unlock() + hasResultAttachments := len(event.Images) > 0 || len(event.Files) > 0 fullResponse := event.Content // When tool progress is hidden, segmentStart stays 0 and textParts // contains ALL text across tool boundaries. Prefer the full accumulated @@ -5294,7 +5301,7 @@ func (e *Engine) processInteractiveEvents(state *interactiveState, session *Sess } else if fullResponse == "" && len(textParts) > 0 { fullResponse = strings.Join(textParts, "") } - if fullResponse == "" { + if fullResponse == "" && !hasResultAttachments { fullResponse = e.i18n.T(MsgEmptyResponse) } @@ -5554,6 +5561,12 @@ func (e *Engine) processInteractiveEvents(state *interactiveState, session *Sess if elapsed := time.Since(replyStart); elapsed >= slowPlatformSend { slog.Warn("slow final reply send", "platform", p.Name(), "elapsed", elapsed, "response_len", len(fullResponse)) } + if !isSilent && hasResultAttachments { + if err := e.sendEventResultAttachments(p, replyCtx, event.Images, event.Files); err != nil { + slog.Error("failed to send EventResult attachments", "platform", p.Name(), "error", err) + e.send(p, replyCtx, fmt.Sprintf(e.i18n.T(MsgError), err)) + } + } // TTS: async voice reply if enabled (skipped for silent replies) if !isSilent && e.tts != nil && e.tts.Enabled && e.tts.TTS != nil { @@ -10736,6 +10749,51 @@ func (e *Engine) SendToSessionWithOptions(sessionKey, message string, images []I return nil } +func (e *Engine) sendEventResultAttachments(p Platform, replyCtx any, images []ImageAttachment, files []FileAttachment) error { + if len(images) == 0 && len(files) == 0 { + return nil + } + if !e.attachmentSendEnabled { + return ErrAttachmentSendDisabled + } + + var imageSender ImageSender + if len(images) > 0 { + var ok bool + imageSender, ok = p.(ImageSender) + if !ok { + return fmt.Errorf("platform %s: %w", p.Name(), ErrNotSupported) + } + } + + var fileSender FileSender + if len(files) > 0 { + var ok bool + fileSender, ok = p.(FileSender) + if !ok { + return fmt.Errorf("platform %s: %w", p.Name(), ErrNotSupported) + } + } + + for _, img := range images { + if err := e.waitOutgoing(p); err != nil { + return err + } + if err := imageSender.SendImage(e.ctx, replyCtx, img); err != nil { + return err + } + } + for _, file := range files { + if err := e.waitOutgoing(p); err != nil { + return err + } + if err := fileSender.SendFile(e.ctx, replyCtx, file); err != nil { + return err + } + } + return nil +} + type sendTarget struct { state *interactiveState platform Platform diff --git a/core/engine_test.go b/core/engine_test.go index 7cd258560..e844c2477 100644 --- a/core/engine_test.go +++ b/core/engine_test.go @@ -1087,6 +1087,39 @@ func TestProcessInteractiveEvents_DoesNotSuppressDifferentFinalText(t *testing.T } } +func TestProcessInteractiveEvents_SendsEventResultAttachments(t *testing.T) { + p := &stubMediaPlatform{stubPlatformEngine: stubPlatformEngine{n: "test"}} + e := NewEngine("test", &stubAgent{}, []Platform{p}, "", LangEnglish) + sessionKey := "test:user1" + session := e.sessions.GetOrCreateActive(sessionKey) + agentSession := newControllableSession("s1") + state := &interactiveState{ + agentSession: agentSession, + platform: p, + replyCtx: "ctx-1", + } + e.interactiveStates[sessionKey] = state + + agentSession.events <- Event{ + Type: EventResult, + Content: "已生成。", + Images: []ImageAttachment{{MimeType: "image/png", Data: []byte("img"), FileName: "generated.png"}}, + Files: []FileAttachment{{MimeType: "text/plain", Data: []byte("doc"), FileName: "report.txt"}}, + Done: true, + } + e.processInteractiveEvents(state, session, e.sessions, sessionKey, "m1", time.Now(), nil, nil, nil) + + if got := p.getSent(); len(got) != 1 || got[0] != "已生成。" { + t.Fatalf("sent text = %#v, want generated text", got) + } + if len(p.images) != 1 || p.images[0].FileName != "generated.png" { + t.Fatalf("images = %#v", p.images) + } + if len(p.files) != 1 || p.files[0].FileName != "report.txt" { + t.Fatalf("files = %#v", p.files) + } +} + // TestProcessInteractiveEvents_NonTerminalResultContinuesTurn pins issue #481: // when Claude Code emits a mid-turn compaction result (Done=false), the engine // must NOT treat it as turn completion. Subsequent EventText (analogous to a @@ -1100,10 +1133,10 @@ func TestProcessInteractiveEvents_NonTerminalResultContinuesTurn(t *testing.T) { session := e.sessions.GetOrCreateActive(sessionKey) agentSession := newControllableSession("s1") state := &interactiveState{ - agentSession: agentSession, - platform: p, - replyCtx: "ctx-1", - currentTurnUserMessageTimeMs: 100, + agentSession: agentSession, + platform: p, + replyCtx: "ctx-1", + currentTurnUserMessageTimeMs: 100, lastCompletedUserMessageTimeMs: 0, } e.interactiveStates[sessionKey] = state @@ -15016,8 +15049,8 @@ func TestIsAllowResponse_WithMultipleMentions(t *testing.T) { func TestIsAllowResponse_NotInsideOtherWord(t *testing.T) { cases := []string{ "禁止允许这种", - "不允许这样", // "不允许" has its own deny entry, but as part of "不允许这样" the user clearly is denying / negating, never allowing. - "我不太允许这件事", // long sentence, no token equals "允许" + "不允许这样", // "不允许" has its own deny entry, but as part of "不允许这样" the user clearly is denying / negating, never allowing. + "我不太允许这件事", // long sentence, no token equals "允许" "please don't allowall the things", // FieldsFunc keeps "allowall" intact, but it is the approveAll single-token form, not allow. "hello world", "", @@ -15045,7 +15078,7 @@ func TestIsDenyResponse_WithMention(t *testing.T) { } negatives := []string{ - "拒绝症患者", // embedded — must not match + "拒绝症患者", // embedded — must not match "我们都不应该 hello", // unrelated } for _, s := range negatives { diff --git a/core/message.go b/core/message.go index b852d9b51..d802605a2 100644 --- a/core/message.go +++ b/core/message.go @@ -231,20 +231,22 @@ type UserQuestionOption struct { // Event represents a single piece of agent output streamed back to the engine. type Event struct { - Type EventType - Content string - ToolName string // populated for EventToolUse, EventPermissionRequest - ToolInput string // human-readable summary of tool input - ToolInputRaw map[string]any // raw tool input (for EventPermissionRequest, used in allow response) - ToolResult string // populated for EventToolResult - ToolStatus string // optional status for EventToolResult (e.g. completed/failed) - ToolExitCode *int // optional exit code for EventToolResult - ToolSuccess *bool // optional success flag for EventToolResult - SessionID string // agent-managed session ID for conversation continuity - RequestID string // unique request ID for EventPermissionRequest - Questions []UserQuestion // populated when ToolName == "AskUserQuestion" - Done bool - Error error + Type EventType + Content string + Images []ImageAttachment // optional images emitted with the event result + Files []FileAttachment // optional files emitted with the event result + ToolName string // populated for EventToolUse, EventPermissionRequest + ToolInput string // human-readable summary of tool input + ToolInputRaw map[string]any // raw tool input (for EventPermissionRequest, used in allow response) + ToolResult string // populated for EventToolResult + ToolStatus string // optional status for EventToolResult (e.g. completed/failed) + ToolExitCode *int // optional exit code for EventToolResult + ToolSuccess *bool // optional success flag for EventToolResult + SessionID string // agent-managed session ID for conversation continuity + RequestID string // unique request ID for EventPermissionRequest + Questions []UserQuestion // populated when ToolName == "AskUserQuestion" + Done bool + Error error InputTokens int // token usage from agent result events OutputTokens int CacheCreationInputTokens int // cache-write tokens (new content written to cache)