From fb96645ea90c265891d845dd61c6c396a781ea70 Mon Sep 17 00:00:00 2001 From: Yiliu Date: Thu, 26 Feb 2026 23:50:33 +0800 Subject: [PATCH 1/7] fix(providers): support lookup-based fallback candidate resolution --- pkg/providers/fallback.go | 17 ++++++++- pkg/providers/fallback_test.go | 69 ++++++++++++++++++++++++++++++++++ 2 files changed, 85 insertions(+), 1 deletion(-) diff --git a/pkg/providers/fallback.go b/pkg/providers/fallback.go index ecd451ec9..7ba563b66 100644 --- a/pkg/providers/fallback.go +++ b/pkg/providers/fallback.go @@ -43,11 +43,26 @@ func NewFallbackChain(cooldown *CooldownTracker) *FallbackChain { // ResolveCandidates parses model config into a deduplicated candidate list. func ResolveCandidates(cfg ModelConfig, defaultProvider string) []FallbackCandidate { + return ResolveCandidatesWithLookup(cfg, defaultProvider, nil) +} + +func ResolveCandidatesWithLookup( + cfg ModelConfig, + defaultProvider string, + lookup func(raw string) (resolved string, ok bool), +) []FallbackCandidate { seen := make(map[string]bool) var candidates []FallbackCandidate addCandidate := func(raw string) { - ref := ParseModelRef(raw, defaultProvider) + candidateRaw := strings.TrimSpace(raw) + if lookup != nil { + if resolved, ok := lookup(candidateRaw); ok { + candidateRaw = resolved + } + } + + ref := ParseModelRef(candidateRaw, defaultProvider) if ref == nil { return } diff --git a/pkg/providers/fallback_test.go b/pkg/providers/fallback_test.go index ebba054ef..1783ebcb5 100644 --- a/pkg/providers/fallback_test.go +++ b/pkg/providers/fallback_test.go @@ -453,6 +453,75 @@ func TestResolveCandidates_EmptyPrimary(t *testing.T) { } } +func TestResolveCandidatesWithLookup_AliasResolvesToNestedModel(t *testing.T) { + cfg := ModelConfig{ + Primary: "step-3.5-flash", + Fallbacks: nil, + } + + lookup := func(raw string) (string, bool) { + if raw == "step-3.5-flash" { + return "openrouter/stepfun/step-3.5-flash:free", true + } + return "", false + } + + candidates := ResolveCandidatesWithLookup(cfg, "", lookup) + if len(candidates) != 1 { + t.Fatalf("candidates = %d, want 1", len(candidates)) + } + if candidates[0].Provider != "openrouter" { + t.Fatalf("provider = %q, want openrouter", candidates[0].Provider) + } + if candidates[0].Model != "stepfun/step-3.5-flash:free" { + t.Fatalf("model = %q, want stepfun/step-3.5-flash:free", candidates[0].Model) + } +} + +func TestResolveCandidatesWithLookup_DeduplicateAfterLookup(t *testing.T) { + cfg := ModelConfig{ + Primary: "step-3.5-flash", + Fallbacks: []string{"openrouter/stepfun/step-3.5-flash:free"}, + } + + lookup := func(raw string) (string, bool) { + if raw == "step-3.5-flash" { + return "openrouter/stepfun/step-3.5-flash:free", true + } + return "", false + } + + candidates := ResolveCandidatesWithLookup(cfg, "", lookup) + if len(candidates) != 1 { + t.Fatalf("candidates = %d, want 1", len(candidates)) + } +} + +func TestResolveCandidatesWithLookup_AliasWithoutProtocolUsesDefaultProvider(t *testing.T) { + cfg := ModelConfig{ + Primary: "glm-5", + Fallbacks: nil, + } + + lookup := func(raw string) (string, bool) { + if raw == "glm-5" { + return "glm-5", true + } + return "", false + } + + candidates := ResolveCandidatesWithLookup(cfg, "openai", lookup) + if len(candidates) != 1 { + t.Fatalf("candidates = %d, want 1", len(candidates)) + } + if candidates[0].Provider != "openai" { + t.Fatalf("provider = %q, want openai", candidates[0].Provider) + } + if candidates[0].Model != "glm-5" { + t.Fatalf("model = %q, want glm-5", candidates[0].Model) + } +} + func TestFallbackExhaustedError_Message(t *testing.T) { e := &FallbackExhaustedError{ Attempts: []FallbackAttempt{ From 3a3862340a765e1fd5a0dd5e64da7eab4c2aa7cf Mon Sep 17 00:00:00 2001 From: Yiliu Date: Thu, 26 Feb 2026 23:50:40 +0800 Subject: [PATCH 2/7] fix(agent): resolve fallback model aliases from model_list --- pkg/agent/instance.go | 42 +++++++++++++++++++++- pkg/agent/instance_test.go | 74 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 115 insertions(+), 1 deletion(-) diff --git a/pkg/agent/instance.go b/pkg/agent/instance.go index a6fd365c7..65a1fe04d 100644 --- a/pkg/agent/instance.go +++ b/pkg/agent/instance.go @@ -92,7 +92,47 @@ func NewAgentInstance( Primary: model, Fallbacks: fallbacks, } - candidates := providers.ResolveCandidates(modelCfg, defaults.Provider) + resolveFromModelList := func(raw string) (string, bool) { + ensureProtocol := func(model string) string { + model = strings.TrimSpace(model) + if model == "" { + return "" + } + if strings.Contains(model, "/") { + return model + } + return "openai/" + model + } + + raw = strings.TrimSpace(raw) + if raw == "" { + return "", false + } + + if cfg != nil { + if mc, err := cfg.GetModelConfig(raw); err == nil && mc != nil && strings.TrimSpace(mc.Model) != "" { + return ensureProtocol(mc.Model), true + } + + for i := range cfg.ModelList { + fullModel := strings.TrimSpace(cfg.ModelList[i].Model) + if fullModel == "" { + continue + } + if fullModel == raw { + return ensureProtocol(fullModel), true + } + _, modelID := providers.ExtractProtocol(fullModel) + if modelID == raw { + return ensureProtocol(fullModel), true + } + } + } + + return "", false + } + + candidates := providers.ResolveCandidatesWithLookup(modelCfg, defaults.Provider, resolveFromModelList) return &AgentInstance{ ID: agentID, diff --git a/pkg/agent/instance_test.go b/pkg/agent/instance_test.go index fcc8e9bea..af1bf2ead 100644 --- a/pkg/agent/instance_test.go +++ b/pkg/agent/instance_test.go @@ -93,3 +93,77 @@ func TestNewAgentInstance_DefaultsTemperatureWhenUnset(t *testing.T) { t.Fatalf("Temperature = %f, want %f", agent.Temperature, 0.7) } } + +func TestNewAgentInstance_ResolveCandidatesFromModelListAlias(t *testing.T) { + tmpDir, err := os.MkdirTemp("", "agent-instance-test-*") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tmpDir) + + cfg := &config.Config{ + Agents: config.AgentsConfig{ + Defaults: config.AgentDefaults{ + Workspace: tmpDir, + Model: "step-3.5-flash", + }, + }, + ModelList: []config.ModelConfig{ + { + ModelName: "step-3.5-flash", + Model: "openrouter/stepfun/step-3.5-flash:free", + APIBase: "https://openrouter.ai/api/v1", + }, + }, + } + + provider := &mockProvider{} + agent := NewAgentInstance(nil, &cfg.Agents.Defaults, cfg, provider) + + if len(agent.Candidates) != 1 { + t.Fatalf("len(Candidates) = %d, want 1", len(agent.Candidates)) + } + if agent.Candidates[0].Provider != "openrouter" { + t.Fatalf("candidate provider = %q, want %q", agent.Candidates[0].Provider, "openrouter") + } + if agent.Candidates[0].Model != "stepfun/step-3.5-flash:free" { + t.Fatalf("candidate model = %q, want %q", agent.Candidates[0].Model, "stepfun/step-3.5-flash:free") + } +} + +func TestNewAgentInstance_ResolveCandidatesFromModelListAliasWithoutProtocol(t *testing.T) { + tmpDir, err := os.MkdirTemp("", "agent-instance-test-*") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tmpDir) + + cfg := &config.Config{ + Agents: config.AgentsConfig{ + Defaults: config.AgentDefaults{ + Workspace: tmpDir, + Model: "glm-5", + }, + }, + ModelList: []config.ModelConfig{ + { + ModelName: "glm-5", + Model: "glm-5", + APIBase: "https://api.z.ai/api/coding/paas/v4", + }, + }, + } + + provider := &mockProvider{} + agent := NewAgentInstance(nil, &cfg.Agents.Defaults, cfg, provider) + + if len(agent.Candidates) != 1 { + t.Fatalf("len(Candidates) = %d, want 1", len(agent.Candidates)) + } + if agent.Candidates[0].Provider != "openai" { + t.Fatalf("candidate provider = %q, want %q", agent.Candidates[0].Provider, "openai") + } + if agent.Candidates[0].Model != "glm-5" { + t.Fatalf("candidate model = %q, want %q", agent.Candidates[0].Model, "glm-5") + } +} From 99582bbd91524c29744c34ed60604b866440c669 Mon Sep 17 00:00:00 2001 From: Yiliu Date: Thu, 26 Feb 2026 23:50:48 +0800 Subject: [PATCH 3/7] docs: add issue 783 investigation and execution plan --- ...issue-783-investigation-and-fix-plan.zh.md | 61 +++++++++++++++++++ 1 file changed, 61 insertions(+) create mode 100644 docs/design/issue-783-investigation-and-fix-plan.zh.md diff --git a/docs/design/issue-783-investigation-and-fix-plan.zh.md b/docs/design/issue-783-investigation-and-fix-plan.zh.md new file mode 100644 index 000000000..1c9fc1e70 --- /dev/null +++ b/docs/design/issue-783-investigation-and-fix-plan.zh.md @@ -0,0 +1,61 @@ +# Issue #783 调研与修复执行文档 + +## 1. 问题澄清（已确认） + +- 现象：当 `agents.*.model.primary/fallbacks` 使用 `model_name` 别名（如 `step-3.5-flash`）时，fallback 链路将别名当作真实 `provider/model` 解析，导致 `provider` 可能为空、`model` 可能错误。 +- 根因：`ResolveCandidates` 仅对字符串做 `ParseModelRef`，未先通过 `model_list` 将别名映射到真实 `model` 字段。 +- 影响： + - fallback 执行可能把别名直接发给 OpenAI-compatible provider，触发 `Unknown Model`。 + - `defaults.provider` 为空时，日志出现 `provider=` 空值。 + +## 2. 本次目标 + +- 修复 fallback 候选解析：优先通过 `model_list` 解析别名。 +- 兼容旧行为：若未命中 `model_list`，继续走原有 `ParseModelRef` 兜底。 +- 补充测试：覆盖别名、嵌套路径模型（如 `openrouter/stepfun/...`）、空默认 provider。 +- 验证代码风格：与当前仓库风格保持一致（命名、错误处理、测试结构）。 + +## 3. 联网最佳实践调研结论（已完成） + +- [x] 查阅 OpenAI-compatible 网关（如 OpenRouter）对 `model` 字段的推荐处理。 +- [x] 查阅多 provider/fallback 设计最佳实践（候选解析、日志可观测性）。 +- [x] 将外部建议映射为本仓库可执行约束。 + +外部参考要点（来自 OpenRouter/LiteLLM/Cloudflare AI Gateway 等官方文档）： + +- 优先显式配置，不依赖字符串切分推断 provider。 +- 对网关模型标识应保留完整路径语义，避免截断导致 Unknown Model。 +- fallback 与 primary 应复用同一解析策略，避免“主路径正确、降级路径错误”。 + +参考链接： + +- OpenRouter Provider Routing: https://openrouter.ai/docs/guides/routing/provider-selection +- OpenRouter Model Fallbacks: https://openrouter.ai/docs/guides/routing/model-fallbacks +- OpenRouter Chat Completion API: https://openrouter.ai/docs/api-reference/chat-completion +- LiteLLM Router Architecture: https://docs.litellm.ai/docs/router_architecture +- Cloudflare AI Gateway Chat Completion: https://developers.cloudflare.com/ai-gateway/usage/chat-completion/ + +与本仓库对应的可执行约束： + +- 在 fallback candidate 构建阶段先做 `model_name -> model_list.model` 映射。 +- 未命中映射时保留旧解析行为，保证兼容性。 +- 用新增测试锁定“别名 + 嵌套模型路径 + 空默认 provider”场景。 + +## 4. 实施步骤（顺序执行） + +- [x] Step 1: 对齐现有代码模式，定位最小改动点（`pkg/agent` + `pkg/providers`）。 +- [x] Step 2: 实现“基于 model_list 的 fallback 候选解析”。 +- [x] Step 3: 增加/更新单元测试，覆盖 issue 场景。 +- [x] Step 4: 代码风格一致性复核（与现有文件风格对照）。 +- [x] Step 5: 运行质量门禁（LSP + `make check`）。 + +## 5. 执行记录 + +- 状态：已完成 +- 已完成改动： + - `pkg/providers/fallback.go`：新增 `ResolveCandidatesWithLookup`，并保持 `ResolveCandidates` 向后兼容。 + - `pkg/agent/instance.go`：在构建 fallback candidates 前，优先通过 `model_list` 解析别名，并对无协议模型补齐默认 `openai/` 前缀后再解析。 + - `pkg/providers/fallback_test.go`：新增别名解析与去重测试。 + - `pkg/agent/instance_test.go`：新增 agent 侧别名解析到嵌套模型路径、无协议模型解析测试。 +- 风格对齐检查（完成）：与 `pkg/providers/fallback_test.go`、`pkg/providers/model_ref_test.go` 现有模式一致。 +- 质量验证（完成）：先 `make generate`，后 `make check` 全量通过。 From 4774f7555b97c8a41280fa655b96ba2f4b55f6ee Mon Sep 17 00:00:00 2001 From: Zhang Rui Date: Fri, 27 Feb 2026 19:24:50 +0800 Subject: [PATCH 4/7] Add WeCom AIBot channel implementation and tests - Introduced WeCom AIBot channel with configuration options in config.go. - Added default values for WeCom AIBot in defaults.go. - Registered WeCom AIBot channel factory in init.go. - Created comprehensive unit tests for WeCom AIBot channel functionalities including creation, start/stop behavior, webhook path handling, message encryption/decryption, and signature generation. --- config/config.example.json | 12 +- pkg/channels/manager.go | 4 + pkg/channels/wecom/aibot.go | 1082 ++++++++++++++++++++++++++++++ pkg/channels/wecom/aibot_test.go | 219 ++++++ pkg/channels/wecom/init.go | 3 + pkg/config/config.go | 39 +- pkg/config/defaults.go | 11 + 7 files changed, 1356 insertions(+), 14 deletions(-) create mode 100644 pkg/channels/wecom/aibot.go create mode 100644 pkg/channels/wecom/aibot_test.go diff --git a/config/config.example.json b/config/config.example.json index 9575039f8..7921b2ef7 100644 --- a/config/config.example.json +++ b/config/config.example.json @@ -115,7 +115,7 @@ "allow_from": [] }, "wecom": { - "_comment": "WeCom Bot (智能机器人) - Easier setup, supports group chats", + "_comment": "WeCom Bot - Easier setup, supports group chats", "enabled": false, "token": "YOUR_TOKEN", "encoding_aes_key": "YOUR_43_CHAR_ENCODING_AES_KEY", @@ -139,6 +139,16 @@ "webhook_path": "/webhook/wecom-app", "allow_from": [], "reply_timeout": 5 + }, + "wecom_aibot": { + "_comment": "WeCom AI Bot (智能机器人) - Official WeCom AI Bot integration, supports proactive messaging and private chats.", + "enabled": false, + "token": "YOUR_TOKEN", + "encoding_aes_key": "YOUR_43_CHAR_ENCODING_AES_KEY", + "webhook_host": "0.0.0.0", + "webhook_port": 18791, + "webhook_path": "/webhook/wecom-aibot", + "max_steps": 10 } }, "providers": { diff --git a/pkg/channels/manager.go b/pkg/channels/manager.go index 38b408f5e..af0baf63c 100644 --- a/pkg/channels/manager.go +++ b/pkg/channels/manager.go @@ -250,6 +250,10 @@ func (m *Manager) initChannels() error { m.initChannel("wecom", "WeCom") } + if m.config.Channels.WeComAIBot.Enabled && m.config.Channels.WeComAIBot.Token != "" { + m.initChannel("wecom_aibot", "WeCom AI Bot") + } + if m.config.Channels.WeComApp.Enabled && m.config.Channels.WeComApp.CorpID != "" { m.initChannel("wecom_app", "WeCom App") } diff --git a/pkg/channels/wecom/aibot.go b/pkg/channels/wecom/aibot.go new file mode 100644 index 000000000..d78f130bd --- /dev/null +++ b/pkg/channels/wecom/aibot.go @@ -0,0 +1,1082 @@ +package wecom + +import ( + "bytes" + "context" + "crypto/aes" + "crypto/cipher" + "crypto/md5" + "crypto/rand" + "crypto/sha1" + "encoding/base64" + "encoding/binary" + "encoding/json" + "fmt" + "io" + "math/big" + "net/http" + "sort" + "strings" + "sync" + "time" + + "github.com/sipeed/picoclaw/pkg/bus" + "github.com/sipeed/picoclaw/pkg/channels" + "github.com/sipeed/picoclaw/pkg/config" + "github.com/sipeed/picoclaw/pkg/logger" + "github.com/sipeed/picoclaw/pkg/utils" +) + +// WeComAIBotChannel implements the Channel interface for WeCom AI Bot (企业微信智能机器人) +type WeComAIBotChannel struct { + *channels.BaseChannel + config config.WeComAIBotConfig + ctx context.Context + cancel context.CancelFunc + streamTasks map[string]*streamTask // streamID -> task (for poll lookups) + chatTasks map[string][]*streamTask // chatID -> in-flight tasks queue (FIFO) + taskMu sync.RWMutex +} + +// streamTask represents a streaming task for AI Bot +type streamTask struct { + StreamID string + ChatID string // used by Send() to find this task + ResponseURL string // temporary URL for proactive reply (valid 1 hour, use once) + Question string + CreatedTime time.Time + Deadline time.Time // ~5m30s, we close the stream here and switch to response_url + StreamClosed bool // stream returned finish:true; waiting for agent to reply via response_url + Finished bool // fully done + mu sync.Mutex + answerCh chan string // receives agent reply from Send() +} + +// WeComAIBotMessage represents the decrypted JSON message from WeCom AI Bot +// Ref: https://developer.work.weixin.qq.com/document/path/100719 +type WeComAIBotMessage struct { + MsgID string `json:"msgid"` + AIBotID string `json:"aibotid"` + ChatID string `json:"chatid"` // only for group chat + ChatType string `json:"chattype"` // "single" or "group" + From struct { + UserID string `json:"userid"` + } `json:"from"` + ResponseURL string `json:"response_url"` // temporary URL for proactive reply + MsgType string `json:"msgtype"` + // text message + Text *struct { + Content string `json:"content"` + } `json:"text,omitempty"` + // stream polling refresh + Stream *struct { + ID string `json:"id"` + } `json:"stream,omitempty"` + // image message + Image *struct { + URL string `json:"url"` + } `json:"image,omitempty"` + // mixed message (text + image) + Mixed *struct { + MsgItem []struct { + MsgType string `json:"msgtype"` + Text *struct { + Content string `json:"content"` + } `json:"text,omitempty"` + Image *struct { + URL string `json:"url"` + } `json:"image,omitempty"` + } `json:"msg_item"` + } `json:"mixed,omitempty"` + // event field + Event string `json:"event,omitempty"` +} + +// WeComAIBotStreamResponse represents the streaming response format +type WeComAIBotStreamResponse struct { + MsgType string `json:"msgtype"` + Stream struct { + ID string `json:"id"` + Finish bool `json:"finish"` + Content string `json:"content,omitempty"` + MsgItem []struct { + MsgType string `json:"msgtype"` + Image *struct { + Base64 string `json:"base64"` + MD5 string `json:"md5"` + } `json:"image,omitempty"` + } `json:"msg_item,omitempty"` + } `json:"stream"` +} + +// WeComAIBotEncryptedResponse represents the encrypted response wrapper +// Fields match WXBizJsonMsgCrypt.generate() in Python SDK +type WeComAIBotEncryptedResponse struct { + Encrypt string `json:"encrypt"` + MsgSignature string `json:"msgsignature"` + Timestamp string `json:"timestamp"` + Nonce string `json:"nonce"` +} + +// NewWeComAIBotChannel creates a new WeCom AI Bot channel instance +func NewWeComAIBotChannel(cfg config.WeComAIBotConfig, messageBus *bus.MessageBus) (*WeComAIBotChannel, error) { + if cfg.Token == "" || cfg.EncodingAESKey == "" { + return nil, fmt.Errorf("token and encoding_aes_key are required for WeCom AI Bot") + } + + base := channels.NewBaseChannel("wecom_aibot", cfg, messageBus, cfg.AllowFrom, + channels.WithMaxMessageLength(2048), + ) + + return &WeComAIBotChannel{ + BaseChannel: base, + config: cfg, + streamTasks: make(map[string]*streamTask), + chatTasks: make(map[string][]*streamTask), + }, nil +} + +// Name returns the channel name +func (c *WeComAIBotChannel) Name() string { + return "wecom_aibot" +} + +// Start initializes the WeCom AI Bot channel +func (c *WeComAIBotChannel) Start(ctx context.Context) error { + logger.InfoC("wecom_aibot", "Starting WeCom AI Bot channel...") + + c.ctx, c.cancel = context.WithCancel(ctx) + + // Start cleanup goroutine for old tasks + go c.cleanupLoop() + + c.SetRunning(true) + logger.InfoC("wecom_aibot", "WeCom AI Bot channel started") + + return nil +} + +// Stop gracefully stops the WeCom AI Bot channel +func (c *WeComAIBotChannel) Stop(ctx context.Context) error { + logger.InfoC("wecom_aibot", "Stopping WeCom AI Bot channel...") + + if c.cancel != nil { + c.cancel() + } + + c.SetRunning(false) + logger.InfoC("wecom_aibot", "WeCom AI Bot channel stopped") + return nil +} + +// Send delivers the agent reply into the active streamTask for msg.ChatID. +// It writes into the earliest unfinished task in the queue (FIFO per chatID). +// If the stream has already closed (deadline passed), it posts directly to response_url. +func (c *WeComAIBotChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { + c.taskMu.Lock() + queue := c.chatTasks[msg.ChatID] + for len(queue) > 0 && queue[0].Finished { + queue = queue[1:] + } + c.chatTasks[msg.ChatID] = queue + var task *streamTask + if len(queue) > 0 { + task = queue[0] + } + c.taskMu.Unlock() + + if task == nil { + logger.DebugCF("wecom_aibot", "Send: no active task for chat (may have timed out)", map[string]any{ + "chat_id": msg.ChatID, + }) + return nil + } + + task.mu.Lock() + streamClosed := task.StreamClosed + responseURL := task.ResponseURL + task.mu.Unlock() + + if streamClosed { + // Stream already ended with a "please wait" notice; send the real reply via response_url. + logger.InfoCF("wecom_aibot", "Sending reply via response_url", map[string]any{ + "stream_id": task.StreamID, + "chat_id": msg.ChatID, + }) + if responseURL != "" { + if err := c.sendViaResponseURL(responseURL, msg.Content); err != nil { + logger.ErrorCF("wecom_aibot", "Failed to send via response_url", map[string]any{ + "error": err, + "stream_id": task.StreamID, + }) + } + } else { + logger.WarnCF("wecom_aibot", "Stream closed but no response_url available", map[string]any{ + "stream_id": task.StreamID, + }) + } + c.removeTask(task) + return nil + } + + // Stream still open: deliver via answerCh for the next poll response. + select { + case task.answerCh <- msg.Content: + case <-ctx.Done(): + return ctx.Err() + } + return nil +} + +// WebhookPath returns the path for registering on the shared HTTP server +func (c *WeComAIBotChannel) WebhookPath() string { + if c.config.WebhookPath == "" { + return "/webhook/wecom-aibot" + } + return c.config.WebhookPath +} + +// ServeHTTP implements http.Handler for the shared HTTP server +func (c *WeComAIBotChannel) ServeHTTP(w http.ResponseWriter, r *http.Request) { + c.handleWebhook(w, r) +} + +// HealthPath returns the health check endpoint path +func (c *WeComAIBotChannel) HealthPath() string { + return c.WebhookPath() + "/health" +} + +// HealthHandler handles health check requests +func (c *WeComAIBotChannel) HealthHandler(w http.ResponseWriter, r *http.Request) { + c.handleHealth(w, r) +} + +// handleWebhook handles incoming webhook requests from WeCom AI Bot +func (c *WeComAIBotChannel) handleWebhook(w http.ResponseWriter, r *http.Request) { + ctx := r.Context() + + // Log all incoming requests for debugging + logger.DebugCF("wecom_aibot", "Received webhook request", map[string]any{ + "method": r.Method, + "path": r.URL.Path, + "query": r.URL.RawQuery, + }) + + if r.Method == http.MethodGet { + // URL verification + c.handleVerification(ctx, w, r) + } else if r.Method == http.MethodPost { + // Message callback + c.handleMessageCallback(ctx, w, r) + } else { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + } +} + +// handleVerification handles the URL verification request from WeCom +func (c *WeComAIBotChannel) handleVerification(ctx context.Context, w http.ResponseWriter, r *http.Request) { + msgSignature := r.URL.Query().Get("msg_signature") + timestamp := r.URL.Query().Get("timestamp") + nonce := r.URL.Query().Get("nonce") + echostr := r.URL.Query().Get("echostr") + + logger.DebugCF("wecom_aibot", "URL verification request", map[string]any{ + "msg_signature": msgSignature, + "timestamp": timestamp, + "nonce": nonce, + }) + + // Verify signature + if !verifySignature(c.config.Token, msgSignature, timestamp, nonce, echostr) { + logger.ErrorC("wecom_aibot", "Signature verification failed") + http.Error(w, "Signature verification failed", http.StatusUnauthorized) + return + } + + // Decrypt echostr + // For WeCom AI Bot (智能机器人), receiveid should be empty string + decrypted, err := decryptMessageWithVerify(echostr, c.config.EncodingAESKey, "") + if err != nil { + logger.ErrorCF("wecom_aibot", "Failed to decrypt echostr", map[string]any{ + "error": err, + }) + http.Error(w, "Decryption failed", http.StatusInternalServerError) + return + } + + // Remove BOM and whitespace as per WeCom documentation + decrypted = strings.TrimPrefix(decrypted, "\ufeff") + decrypted = strings.TrimSpace(decrypted) + + logger.InfoC("wecom_aibot", "URL verification successful") + w.Header().Set("Content-Type", "text/plain; charset=utf-8") + w.WriteHeader(http.StatusOK) + w.Write([]byte(decrypted)) +} + +// handleMessageCallback handles incoming messages from WeCom AI Bot +func (c *WeComAIBotChannel) handleMessageCallback(ctx context.Context, w http.ResponseWriter, r *http.Request) { + msgSignature := r.URL.Query().Get("msg_signature") + timestamp := r.URL.Query().Get("timestamp") + nonce := r.URL.Query().Get("nonce") + + // Read request body + body, err := io.ReadAll(r.Body) + if err != nil { + logger.ErrorCF("wecom_aibot", "Failed to read request body", map[string]any{ + "error": err, + }) + http.Error(w, "Failed to read body", http.StatusBadRequest) + return + } + + // Parse JSON body to get encrypted message + // Format: {"encrypt": "base64_encrypted_string"} + var encryptedMsg struct { + Encrypt string `json:"encrypt"` + } + if err := json.Unmarshal(body, &encryptedMsg); err != nil { + logger.ErrorCF("wecom_aibot", "Failed to parse JSON body", map[string]any{ + "error": err, + "body": string(body), + }) + http.Error(w, "Failed to parse JSON", http.StatusBadRequest) + return + } + + // Verify signature + if !verifySignature(c.config.Token, msgSignature, timestamp, nonce, encryptedMsg.Encrypt) { + logger.ErrorC("wecom_aibot", "Signature verification failed") + http.Error(w, "Signature verification failed", http.StatusUnauthorized) + return + } + + // Decrypt message + // For WeCom AI Bot (智能机器人), receiveid is empty string + decrypted, err := decryptMessageWithVerify(encryptedMsg.Encrypt, c.config.EncodingAESKey, "") + if err != nil { + logger.ErrorCF("wecom_aibot", "Failed to decrypt message", map[string]any{ + "error": err, + }) + http.Error(w, "Decryption failed", http.StatusInternalServerError) + return + } + + // Parse decrypted JSON message + var msg WeComAIBotMessage + if err := json.Unmarshal([]byte(decrypted), &msg); err != nil { + logger.ErrorCF("wecom_aibot", "Failed to parse decrypted JSON", map[string]any{ + "error": err, + "decrypted": decrypted, + }) + http.Error(w, "Failed to parse message", http.StatusInternalServerError) + return + } + + logger.DebugCF("wecom_aibot", "Decrypted message", map[string]any{ + "msgtype": msg.MsgType, + }) + + // Process the message and get streaming response + response := c.processMessage(ctx, msg, timestamp, nonce) + + // Return encrypted JSON response + w.Header().Set("Content-Type", "application/json; charset=utf-8") + w.WriteHeader(http.StatusOK) + w.Write([]byte(response)) +} + +// processMessage processes the received message and returns encrypted response +func (c *WeComAIBotChannel) processMessage(ctx context.Context, msg WeComAIBotMessage, timestamp, nonce string) string { + logger.DebugCF("wecom_aibot", "Processing message", map[string]any{ + "msgtype": msg.MsgType, + }) + + switch msg.MsgType { + case "text": + return c.handleTextMessage(ctx, msg, timestamp, nonce) + case "stream": + return c.handleStreamMessage(ctx, msg, timestamp, nonce) + case "image": + return c.handleImageMessage(ctx, msg, timestamp, nonce) + case "mixed": + return c.handleMixedMessage(ctx, msg, timestamp, nonce) + case "event": + return c.handleEventMessage(ctx, msg, timestamp, nonce) + default: + logger.WarnCF("wecom_aibot", "Unsupported message type", map[string]any{ + "msgtype": msg.MsgType, + }) + return c.encryptResponse("", timestamp, nonce, WeComAIBotStreamResponse{ + MsgType: "stream", + Stream: struct { + ID string `json:"id"` + Finish bool `json:"finish"` + Content string `json:"content,omitempty"` + MsgItem []struct { + MsgType string `json:"msgtype"` + Image *struct { + Base64 string `json:"base64"` + MD5 string `json:"md5"` + } `json:"image,omitempty"` + } `json:"msg_item,omitempty"` + }{ + ID: c.generateStreamID(), + Finish: true, + Content: "暂不支持该消息类型", + }, + }) + } +} + +// handleTextMessage handles text messages by starting a new streaming task +func (c *WeComAIBotChannel) handleTextMessage(ctx context.Context, msg WeComAIBotMessage, timestamp, nonce string) string { + if msg.Text == nil { + logger.ErrorC("wecom_aibot", "text message missing text field") + return c.encryptEmptyResponse(timestamp, nonce) + } + + content := msg.Text.Content + userID := msg.From.UserID + if userID == "" { + userID = "unknown" + } + + // chatID: group chat uses chatid, single chat uses userid + chatID := msg.ChatID + if chatID == "" { + chatID = userID + } + + streamID := c.generateStreamID() + + // WeCom stops sending stream-refresh callbacks after 6 minutes. + // Set a slightly shorter deadline so we can send a timeout notice before it gives up. + deadline := time.Now().Add(30 * time.Second) + + task := &streamTask{ + StreamID: streamID, + ChatID: chatID, + ResponseURL: msg.ResponseURL, + Question: content, + CreatedTime: time.Now(), + Deadline: deadline, + Finished: false, + answerCh: make(chan string, 1), + } + + c.taskMu.Lock() + c.streamTasks[streamID] = task + c.chatTasks[chatID] = append(c.chatTasks[chatID], task) + c.taskMu.Unlock() + + // Publish to agent asynchronously; agent will call Send() with reply + // Use c.ctx (channel lifetime) instead of r.Context() which is canceled when the HTTP handler returns. + go func() { + sender := bus.SenderInfo{ + Platform: "wecom_aibot", + PlatformID: userID, + CanonicalID: "wecom_aibot:" + userID, + DisplayName: userID, + } + peerKind := "direct" + if msg.ChatType == "group" { + peerKind = "group" + } + peer := bus.Peer{Kind: peerKind, ID: chatID} + metadata := map[string]string{ + "channel": "wecom_aibot", + "chat_type": msg.ChatType, + "msg_type": "text", + "msgid": msg.MsgID, + "aibotid": msg.AIBotID, + "stream_id": streamID, + "response_url": msg.ResponseURL, + } + c.HandleMessage(c.ctx, peer, msg.MsgID, userID, chatID, + content, nil, metadata, sender) + }() + + // Return first streaming response immediately (finish=false, content empty) + return c.getStreamResponse(task, timestamp, nonce) +} + +// handleStreamMessage handles stream polling requests +func (c *WeComAIBotChannel) handleStreamMessage(ctx context.Context, msg WeComAIBotMessage, timestamp, nonce string) string { + if msg.Stream == nil { + logger.ErrorC("wecom_aibot", "Stream message missing stream field") + return c.encryptEmptyResponse(timestamp, nonce) + } + + streamID := msg.Stream.ID + + c.taskMu.RLock() + task, exists := c.streamTasks[streamID] + c.taskMu.RUnlock() + + if !exists { + logger.DebugCF("wecom_aibot", "Stream task not found (may be from previous session)", map[string]any{ + "stream_id": streamID, + }) + return c.encryptResponse(streamID, timestamp, nonce, WeComAIBotStreamResponse{ + MsgType: "stream", + Stream: struct { + ID string `json:"id"` + Finish bool `json:"finish"` + Content string `json:"content,omitempty"` + MsgItem []struct { + MsgType string `json:"msgtype"` + Image *struct { + Base64 string `json:"base64"` + MD5 string `json:"md5"` + } `json:"image,omitempty"` + } `json:"msg_item,omitempty"` + }{ + ID: streamID, + Finish: true, + Content: "任务不存在或已过期", + }, + }) + } + + // Get next response + return c.getStreamResponse(task, timestamp, nonce) +} + +// handleImageMessage handles image messages +func (c *WeComAIBotChannel) handleImageMessage(ctx context.Context, msg WeComAIBotMessage, timestamp, nonce string) string { + if msg.Image == nil { + logger.ErrorC("wecom_aibot", "Image message missing image field") + return c.encryptEmptyResponse(timestamp, nonce) + } + + imageURL := msg.Image.URL + + // Download and decrypt image + imageData, err := c.downloadAndDecryptImage(ctx, imageURL) + if err != nil { + logger.ErrorCF("wecom_aibot", "Failed to process image", map[string]any{ + "error": err, + "url": imageURL, + }) + return c.encryptResponse("", timestamp, nonce, WeComAIBotStreamResponse{ + MsgType: "stream", + Stream: struct { + ID string `json:"id"` + Finish bool `json:"finish"` + Content string `json:"content,omitempty"` + MsgItem []struct { + MsgType string `json:"msgtype"` + Image *struct { + Base64 string `json:"base64"` + MD5 string `json:"md5"` + } `json:"image,omitempty"` + } `json:"msg_item,omitempty"` + }{ + ID: c.generateStreamID(), + Finish: true, + Content: fmt.Sprintf("图片处理失败: %v", err), + }, + }) + } + + // Echo back the image (simple demo behavior) + streamID := c.generateStreamID() + return c.encryptImageResponse(streamID, timestamp, nonce, imageData) +} + +// handleMixedMessage handles mixed (text + image) messages +func (c *WeComAIBotChannel) handleMixedMessage(ctx context.Context, msg WeComAIBotMessage, timestamp, nonce string) string { + logger.WarnC("wecom_aibot", "Mixed message type not yet fully implemented") + return c.encryptResponse("", timestamp, nonce, WeComAIBotStreamResponse{ + MsgType: "stream", + Stream: struct { + ID string `json:"id"` + Finish bool `json:"finish"` + Content string `json:"content,omitempty"` + MsgItem []struct { + MsgType string `json:"msgtype"` + Image *struct { + Base64 string `json:"base64"` + MD5 string `json:"md5"` + } `json:"image,omitempty"` + } `json:"msg_item,omitempty"` + }{ + ID: c.generateStreamID(), + Finish: true, + Content: "暂不支持图文混排消息", + }, + }) +} + +// handleEventMessage handles event messages +func (c *WeComAIBotChannel) handleEventMessage(ctx context.Context, msg WeComAIBotMessage, timestamp, nonce string) string { + logger.DebugCF("wecom_aibot", "Received event", map[string]any{ + "event": msg.Event, + }) + // Return empty response for events + return c.encryptEmptyResponse(timestamp, nonce) +} + +// getStreamResponse gets the next streaming response for a task. +// - If agent replied: return finish=true with the real answer. +// - If deadline passed: return finish=true with a "please wait" notice, keep task alive for response_url. +// - Otherwise: return finish=false (empty), client will poll again. +func (c *WeComAIBotChannel) getStreamResponse(task *streamTask, timestamp, nonce string) string { + var content string + var finish bool + var closeStreamOnly bool // close stream but do NOT remove task (response_url still pending) + + select { + case answer := <-task.answerCh: + // Agent replied before deadline — normal finish. + content = answer + finish = true + default: + if time.Now().After(task.Deadline) { + // Deadline reached: close the stream with a notice, then wait for agent via response_url. + content = "⏳ Processing, please wait. The results will be sent shortly." + finish = true + closeStreamOnly = true + logger.InfoCF("wecom_aibot", "Stream deadline reached, switching to response_url mode", map[string]any{ + "stream_id": task.StreamID, + "chat_id": task.ChatID, + "response_url": task.ResponseURL != "", + }) + } + // else: still waiting, return finish=false + } + + if finish && !closeStreamOnly { + // Normal finish: remove from all maps. + c.removeTask(task) + } else if closeStreamOnly { + // Only mark stream as closed; keep in chatTasks for Send() to find. + task.mu.Lock() + task.StreamClosed = true + task.mu.Unlock() + // Remove from streamTasks (no more stream polls expected). + c.taskMu.Lock() + delete(c.streamTasks, task.StreamID) + c.taskMu.Unlock() + } + + response := WeComAIBotStreamResponse{ + MsgType: "stream", + Stream: struct { + ID string `json:"id"` + Finish bool `json:"finish"` + Content string `json:"content,omitempty"` + MsgItem []struct { + MsgType string `json:"msgtype"` + Image *struct { + Base64 string `json:"base64"` + MD5 string `json:"md5"` + } `json:"image,omitempty"` + } `json:"msg_item,omitempty"` + }{ + ID: task.StreamID, + Finish: finish, + Content: content, + }, + } + + return c.encryptResponse(task.StreamID, timestamp, nonce, response) +} + +// removeTask removes a task from both streamTasks and chatTasks and marks it finished. +func (c *WeComAIBotChannel) removeTask(task *streamTask) { + task.mu.Lock() + task.Finished = true + task.mu.Unlock() + + c.taskMu.Lock() + delete(c.streamTasks, task.StreamID) + queue := c.chatTasks[task.ChatID] + for i, t := range queue { + if t == task { + c.chatTasks[task.ChatID] = append(queue[:i], queue[i+1:]...) + break + } + } + if len(c.chatTasks[task.ChatID]) == 0 { + delete(c.chatTasks, task.ChatID) + } + c.taskMu.Unlock() +} + +// sendViaResponseURL posts a markdown reply to the WeCom response_url. +// response_url is valid for 1 hour and can only be used once per callback. +func (c *WeComAIBotChannel) sendViaResponseURL(responseURL, content string) error { + payload := map[string]any{ + "msgtype": "markdown", + "markdown": map[string]string{ + "content": content, + }, + } + body, err := json.Marshal(payload) + if err != nil { + return fmt.Errorf("failed to marshal payload: %w", err) + } + + ctx, cancel := context.WithTimeout(c.ctx, 15*time.Second) + defer cancel() + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, responseURL, bytes.NewBuffer(body)) + if err != nil { + return fmt.Errorf("failed to create request: %w", err) + } + req.Header.Set("Content-Type", "application/json; charset=utf-8") + + client := &http.Client{Timeout: 15 * time.Second} + resp, err := client.Do(req) + if err != nil { + return fmt.Errorf("failed to post to response_url: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + respBody, _ := io.ReadAll(resp.Body) + return fmt.Errorf("response_url returned %d: %s", resp.StatusCode, string(respBody)) + } + return nil +} + +// encryptResponse encrypts a streaming response +func (c *WeComAIBotChannel) encryptResponse(streamID, timestamp, nonce string, response WeComAIBotStreamResponse) string { + // Marshal response to JSON + plaintext, err := json.Marshal(response) + if err != nil { + logger.ErrorCF("wecom_aibot", "Failed to marshal response", map[string]any{ + "error": err, + }) + return "" + } + + logger.DebugCF("wecom_aibot", "Encrypting response", map[string]any{ + "stream_id": streamID, + "finish": response.Stream.Finish, + "preview": utils.Truncate(response.Stream.Content, 100), + }) + + // Encrypt message + encrypted, err := c.encryptMessage(string(plaintext), "") + if err != nil { + logger.ErrorCF("wecom_aibot", "Failed to encrypt message", map[string]any{ + "error": err, + }) + return "" + } + + // Generate signature + signature := c.generateSignature(timestamp, nonce, encrypted) + + // Build encrypted response + encryptedResp := WeComAIBotEncryptedResponse{ + Encrypt: encrypted, + MsgSignature: signature, + Timestamp: timestamp, + Nonce: nonce, + } + + respJSON, err := json.Marshal(encryptedResp) + if err != nil { + logger.ErrorCF("wecom_aibot", "Failed to marshal encrypted response", map[string]any{ + "error": err, + }) + return "" + } + + logger.DebugCF("wecom_aibot", "Response encrypted", map[string]any{ + "stream_id": streamID, + }) + + return string(respJSON) +} + +// encryptImageResponse encrypts an image response +func (c *WeComAIBotChannel) encryptImageResponse(streamID, timestamp, nonce string, imageData []byte) string { + // Calculate MD5 + hash := md5.Sum(imageData) + imageMD5 := fmt.Sprintf("%x", hash) + + // Encode to base64 + imageBase64 := base64.StdEncoding.EncodeToString(imageData) + + response := WeComAIBotStreamResponse{ + MsgType: "stream", + Stream: struct { + ID string `json:"id"` + Finish bool `json:"finish"` + Content string `json:"content,omitempty"` + MsgItem []struct { + MsgType string `json:"msgtype"` + Image *struct { + Base64 string `json:"base64"` + MD5 string `json:"md5"` + } `json:"image,omitempty"` + } `json:"msg_item,omitempty"` + }{ + ID: streamID, + Finish: true, + MsgItem: []struct { + MsgType string `json:"msgtype"` + Image *struct { + Base64 string `json:"base64"` + MD5 string `json:"md5"` + } `json:"image,omitempty"` + }{ + { + MsgType: "image", + Image: &struct { + Base64 string `json:"base64"` + MD5 string `json:"md5"` + }{ + Base64: imageBase64, + MD5: imageMD5, + }, + }, + }, + }, + } + + return c.encryptResponse(streamID, timestamp, nonce, response) +} + +// encryptEmptyResponse returns empty encrypted response +func (c *WeComAIBotChannel) encryptEmptyResponse(timestamp, nonce string) string { + return "" +} + +// encryptMessage encrypts a plain text message for WeCom AI Bot +func (c *WeComAIBotChannel) encryptMessage(plaintext, receiveid string) (string, error) { + // Decode AES key + aesKey, err := base64.StdEncoding.DecodeString(c.config.EncodingAESKey + "=") + if err != nil { + return "", fmt.Errorf("failed to decode AES key: %w", err) + } + + if len(aesKey) != 32 { + return "", fmt.Errorf("invalid AES key length: %d", len(aesKey)) + } + + // Generate 16-byte random string + randomBytes := make([]byte, 16) + for i := 0; i < 16; i++ { + n, err := rand.Int(rand.Reader, big.NewInt(10)) + if err != nil { + return "", fmt.Errorf("failed to generate random: %w", err) + } + randomBytes[i] = byte('0' + n.Int64()) + } + + // Build message: random(16) + msg_len(4) + msg + receiveid + plaintextBytes := []byte(plaintext) + receiveidBytes := []byte(receiveid) + + msgLen := uint32(len(plaintextBytes)) + msgLenBytes := make([]byte, 4) + binary.BigEndian.PutUint32(msgLenBytes, msgLen) + + // Concatenate + var buffer bytes.Buffer + buffer.Write(randomBytes) + buffer.Write(msgLenBytes) + buffer.Write(plaintextBytes) + buffer.Write(receiveidBytes) + + // PKCS7 padding + plainData := buffer.Bytes() + plainData = pkcs7Pad(plainData, blockSize) + + // AES-CBC encrypt + block, err := aes.NewCipher(aesKey) + if err != nil { + return "", fmt.Errorf("failed to create cipher: %w", err) + } + + ciphertext := make([]byte, len(plainData)) + iv := aesKey[:aes.BlockSize] + mode := cipher.NewCBCEncrypter(block, iv) + mode.CryptBlocks(ciphertext, plainData) + + // Base64 encode + encoded := base64.StdEncoding.EncodeToString(ciphertext) + + return encoded, nil +} + +// pkcs7Pad adds PKCS7 padding +func pkcs7Pad(data []byte, blockSize int) []byte { + padding := blockSize - (len(data) % blockSize) + if padding == 0 { + padding = blockSize + } + padText := bytes.Repeat([]byte{byte(padding)}, padding) + return append(data, padText...) +} + +// generateSignature generates message signature using common function +func (c *WeComAIBotChannel) generateSignature(timestamp, nonce, encrypt string) string { + // Sort parameters + params := []string{c.config.Token, timestamp, nonce, encrypt} + sort.Strings(params) + + // Concatenate + str := strings.Join(params, "") + + // SHA1 hash + hash := sha1.Sum([]byte(str)) + return fmt.Sprintf("%x", hash) +} + +// generateStreamID generates a random stream ID +func (c *WeComAIBotChannel) generateStreamID() string { + const letters = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789" + b := make([]byte, 10) + for i := range b { + n, _ := rand.Int(rand.Reader, big.NewInt(int64(len(letters)))) + b[i] = letters[n.Int64()] + } + return string(b) +} + +// downloadAndDecryptImage downloads and decrypts an encrypted image +func (c *WeComAIBotChannel) downloadAndDecryptImage(ctx context.Context, imageURL string) ([]byte, error) { + // Download image + req, err := http.NewRequestWithContext(ctx, http.MethodGet, imageURL, nil) + if err != nil { + return nil, fmt.Errorf("failed to create request: %w", err) + } + + client := &http.Client{ + Timeout: 15 * time.Second, + } + + resp, err := client.Do(req) + if err != nil { + return nil, fmt.Errorf("failed to download image: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("download failed with status: %d", resp.StatusCode) + } + + encryptedData, err := io.ReadAll(resp.Body) + if err != nil { + return nil, fmt.Errorf("failed to read image data: %w", err) + } + + logger.DebugCF("wecom_aibot", "Image downloaded", map[string]any{ + "size": len(encryptedData), + }) + + // Decode AES key + aesKey, err := base64.StdEncoding.DecodeString(c.config.EncodingAESKey + "=") + if err != nil { + return nil, fmt.Errorf("failed to decode AES key: %w", err) + } + + if len(aesKey) != 32 { + return nil, fmt.Errorf("invalid AES key length: %d", len(aesKey)) + } + + // Decrypt image (AES-CBC) + block, err := aes.NewCipher(aesKey) + if err != nil { + return nil, fmt.Errorf("failed to create cipher: %w", err) + } + + if len(encryptedData)%aes.BlockSize != 0 { + return nil, fmt.Errorf("encrypted data size not multiple of block size") + } + + iv := aesKey[:aes.BlockSize] + mode := cipher.NewCBCDecrypter(block, iv) + + decryptedData := make([]byte, len(encryptedData)) + mode.CryptBlocks(decryptedData, encryptedData) + + // Remove PKCS7 padding + decryptedData, err = pkcs7Unpad(decryptedData) + if err != nil { + return nil, fmt.Errorf("failed to unpad: %w", err) + } + + logger.DebugCF("wecom_aibot", "Image decrypted", map[string]any{ + "size": len(decryptedData), + }) + + return decryptedData, nil +} + +// cleanupLoop periodically cleans up old streaming tasks +func (c *WeComAIBotChannel) cleanupLoop() { + ticker := time.NewTicker(5 * time.Minute) + defer ticker.Stop() + + for { + select { + case <-ticker.C: + c.cleanupOldTasks() + case <-c.ctx.Done(): + return + } + } +} + +// cleanupOldTasks removes tasks that have been alive longer than 1 hour +// (response_url validity window), which is the absolute maximum lifetime of any task. +func (c *WeComAIBotChannel) cleanupOldTasks() { + c.taskMu.Lock() + defer c.taskMu.Unlock() + + cutoff := time.Now().Add(-1 * time.Hour) + for id, task := range c.streamTasks { + if task.CreatedTime.Before(cutoff) { + delete(c.streamTasks, id) + queue := c.chatTasks[task.ChatID] + for i, t := range queue { + if t == task { + c.chatTasks[task.ChatID] = append(queue[:i], queue[i+1:]...) + break + } + } + if len(c.chatTasks[task.ChatID]) == 0 { + delete(c.chatTasks, task.ChatID) + } + logger.DebugCF("wecom_aibot", "Cleaned up expired task", map[string]any{ + "stream_id": id, + }) + } + } + // Also clean up StreamClosed tasks from chatTasks that are older than 1 hour. + for chatID, queue := range c.chatTasks { + filtered := queue[:0] + for _, t := range queue { + if !t.Finished && t.CreatedTime.After(cutoff) { + filtered = append(filtered, t) + } + } + if len(filtered) == 0 { + delete(c.chatTasks, chatID) + } else { + c.chatTasks[chatID] = filtered + } + } +} + +// handleHealth handles health check requests +func (c *WeComAIBotChannel) handleHealth(w http.ResponseWriter, r *http.Request) { + status := "ok" + if !c.IsRunning() { + status = "not running" + } + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + json.NewEncoder(w).Encode(map[string]string{ + "status": status, + }) +} diff --git a/pkg/channels/wecom/aibot_test.go b/pkg/channels/wecom/aibot_test.go new file mode 100644 index 000000000..d15cf5374 --- /dev/null +++ b/pkg/channels/wecom/aibot_test.go @@ -0,0 +1,219 @@ +package wecom + +import ( + "context" + "testing" + + "github.com/sipeed/picoclaw/pkg/bus" + "github.com/sipeed/picoclaw/pkg/config" +) + +func TestNewWeComAIBotChannel(t *testing.T) { + t.Run("success with valid config", func(t *testing.T) { + cfg := config.WeComAIBotConfig{ + Enabled: true, + Token: "test_token", + EncodingAESKey: "testkey1234567890123456789012345678901234567", + WebhookPath: "/webhook/test", + } + + messageBus := bus.NewMessageBus() + ch, err := NewWeComAIBotChannel(cfg, messageBus) + + if err != nil { + t.Fatalf("Expected no error, got %v", err) + } + + if ch == nil { + t.Fatal("Expected channel to be created") + } + + if ch.Name() != "wecom_aibot" { + t.Errorf("Expected name 'wecom_aibot', got '%s'", ch.Name()) + } + }) + + t.Run("error with missing token", func(t *testing.T) { + cfg := config.WeComAIBotConfig{ + Enabled: true, + EncodingAESKey: "testkey1234567890123456789012345678901234567", + } + + messageBus := bus.NewMessageBus() + _, err := NewWeComAIBotChannel(cfg, messageBus) + + if err == nil { + t.Fatal("Expected error for missing token, got nil") + } + }) + + t.Run("error with missing encoding key", func(t *testing.T) { + cfg := config.WeComAIBotConfig{ + Enabled: true, + Token: "test_token", + } + + messageBus := bus.NewMessageBus() + _, err := NewWeComAIBotChannel(cfg, messageBus) + + if err == nil { + t.Fatal("Expected error for missing encoding key, got nil") + } + }) +} + +func TestWeComAIBotChannelStartStop(t *testing.T) { + cfg := config.WeComAIBotConfig{ + Enabled: true, + Token: "test_token", + EncodingAESKey: "testkey1234567890123456789012345678901234567", + } + + messageBus := bus.NewMessageBus() + ch, err := NewWeComAIBotChannel(cfg, messageBus) + if err != nil { + t.Fatalf("Failed to create channel: %v", err) + } + + ctx := context.Background() + + // Test Start + if err := ch.Start(ctx); err != nil { + t.Fatalf("Failed to start channel: %v", err) + } + + if !ch.IsRunning() { + t.Error("Expected channel to be running") + } + + // Test Stop + if err := ch.Stop(ctx); err != nil { + t.Fatalf("Failed to stop channel: %v", err) + } + + if ch.IsRunning() { + t.Error("Expected channel to be stopped") + } +} + +func TestWeComAIBotChannelWebhookPath(t *testing.T) { + t.Run("default path", func(t *testing.T) { + cfg := config.WeComAIBotConfig{ + Enabled: true, + Token: "test_token", + EncodingAESKey: "testkey1234567890123456789012345678901234567", + } + + messageBus := bus.NewMessageBus() + ch, _ := NewWeComAIBotChannel(cfg, messageBus) + + expectedPath := "/webhook/wecom-aibot" + if ch.WebhookPath() != expectedPath { + t.Errorf("Expected webhook path '%s', got '%s'", expectedPath, ch.WebhookPath()) + } + }) + + t.Run("custom path", func(t *testing.T) { + customPath := "/custom/webhook" + cfg := config.WeComAIBotConfig{ + Enabled: true, + Token: "test_token", + EncodingAESKey: "testkey1234567890123456789012345678901234567", + WebhookPath: customPath, + } + + messageBus := bus.NewMessageBus() + ch, _ := NewWeComAIBotChannel(cfg, messageBus) + + if ch.WebhookPath() != customPath { + t.Errorf("Expected webhook path '%s', got '%s'", customPath, ch.WebhookPath()) + } + }) +} + +func TestGenerateStreamID(t *testing.T) { + cfg := config.WeComAIBotConfig{ + Enabled: true, + Token: "test_token", + EncodingAESKey: "testkey1234567890123456789012345678901234567", + } + + messageBus := bus.NewMessageBus() + ch, _ := NewWeComAIBotChannel(cfg, messageBus) + + // Generate multiple IDs and check they are unique + ids := make(map[string]bool) + for i := 0; i < 100; i++ { + id := ch.generateStreamID() + + if len(id) != 10 { + t.Errorf("Expected stream ID length 10, got %d", len(id)) + } + + if ids[id] { + t.Errorf("Duplicate stream ID generated: %s", id) + } + ids[id] = true + } +} + +func TestEncryptDecrypt(t *testing.T) { + // Use a valid 43-character base64 key (企业微信标准格式) + cfg := config.WeComAIBotConfig{ + Enabled: true, + Token: "test_token", + EncodingAESKey: "abcdefghijklmnopqrstuvwxyz0123456789ABCDEFG", // 43 characters + } + + messageBus := bus.NewMessageBus() + ch, _ := NewWeComAIBotChannel(cfg, messageBus) + + plaintext := "Hello, World!" + receiveid := "" + + // Encrypt + encrypted, err := ch.encryptMessage(plaintext, receiveid) + if err != nil { + t.Fatalf("Failed to encrypt message: %v", err) + } + + if encrypted == "" { + t.Fatal("Encrypted message is empty") + } + + // Decrypt + decrypted, err := decryptMessageWithVerify(encrypted, cfg.EncodingAESKey, receiveid) + if err != nil { + t.Fatalf("Failed to decrypt message: %v", err) + } + + if decrypted != plaintext { + t.Errorf("Expected decrypted message '%s', got '%s'", plaintext, decrypted) + } +} + +func TestGenerateSignature(t *testing.T) { + cfg := config.WeComAIBotConfig{ + Enabled: true, + Token: "test_token", + EncodingAESKey: "testkey1234567890123456789012345678901234567", + } + + messageBus := bus.NewMessageBus() + ch, _ := NewWeComAIBotChannel(cfg, messageBus) + + timestamp := "1234567890" + nonce := "test_nonce" + encrypt := "encrypted_msg" + + signature := ch.generateSignature(timestamp, nonce, encrypt) + + if signature == "" { + t.Error("Generated signature is empty") + } + + // Verify signature using verifySignature function + if !verifySignature(cfg.Token, signature, timestamp, nonce, encrypt) { + t.Error("Generated signature does not verify correctly") + } +} diff --git a/pkg/channels/wecom/init.go b/pkg/channels/wecom/init.go index 3ef1ecdf3..bc5a70fa3 100644 --- a/pkg/channels/wecom/init.go +++ b/pkg/channels/wecom/init.go @@ -13,4 +13,7 @@ func init() { channels.RegisterFactory("wecom_app", func(cfg *config.Config, b *bus.MessageBus) (channels.Channel, error) { return NewWeComAppChannel(cfg.Channels.WeComApp, b) }) + channels.RegisterFactory("wecom_aibot", func(cfg *config.Config, b *bus.MessageBus) (channels.Channel, error) { + return NewWeComAIBotChannel(cfg.Channels.WeComAIBot, b) + }) } diff --git a/pkg/config/config.go b/pkg/config/config.go index bdd4d8823..dce3f5de3 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -190,19 +190,20 @@ func (d *AgentDefaults) GetModelName() string { } type ChannelsConfig struct { - WhatsApp WhatsAppConfig `json:"whatsapp"` - Telegram TelegramConfig `json:"telegram"` - Feishu FeishuConfig `json:"feishu"` - Discord DiscordConfig `json:"discord"` - MaixCam MaixCamConfig `json:"maixcam"` - QQ QQConfig `json:"qq"` - DingTalk DingTalkConfig `json:"dingtalk"` - Slack SlackConfig `json:"slack"` - LINE LINEConfig `json:"line"` - OneBot OneBotConfig `json:"onebot"` - WeCom WeComConfig `json:"wecom"` - WeComApp WeComAppConfig `json:"wecom_app"` - Pico PicoConfig `json:"pico"` + WhatsApp WhatsAppConfig `json:"whatsapp"` + Telegram TelegramConfig `json:"telegram"` + Feishu FeishuConfig `json:"feishu"` + Discord DiscordConfig `json:"discord"` + MaixCam MaixCamConfig `json:"maixcam"` + QQ QQConfig `json:"qq"` + DingTalk DingTalkConfig `json:"dingtalk"` + Slack SlackConfig `json:"slack"` + LINE LINEConfig `json:"line"` + OneBot OneBotConfig `json:"onebot"` + WeCom WeComConfig `json:"wecom"` + WeComApp WeComAppConfig `json:"wecom_app"` + WeComAIBot WeComAIBotConfig `json:"wecom_aibot"` + Pico PicoConfig `json:"pico"` } // GroupTriggerConfig controls when the bot responds in group chats. @@ -341,6 +342,18 @@ type WeComAppConfig struct { GroupTrigger GroupTriggerConfig `json:"group_trigger,omitempty"` } +type WeComAIBotConfig struct { + Enabled bool `json:"enabled" env:"PICOCLAW_CHANNELS_WECOM_AIBOT_ENABLED"` + Token string `json:"token" env:"PICOCLAW_CHANNELS_WECOM_AIBOT_TOKEN"` + EncodingAESKey string `json:"encoding_aes_key" env:"PICOCLAW_CHANNELS_WECOM_AIBOT_ENCODING_AES_KEY"` + WebhookHost string `json:"webhook_host" env:"PICOCLAW_CHANNELS_WECOM_AIBOT_WEBHOOK_HOST"` + WebhookPort int `json:"webhook_port" env:"PICOCLAW_CHANNELS_WECOM_AIBOT_WEBHOOK_PORT"` + WebhookPath string `json:"webhook_path" env:"PICOCLAW_CHANNELS_WECOM_AIBOT_WEBHOOK_PATH"` + AllowFrom FlexibleStringSlice `json:"allow_from" env:"PICOCLAW_CHANNELS_WECOM_AIBOT_ALLOW_FROM"` + ReplyTimeout int `json:"reply_timeout" env:"PICOCLAW_CHANNELS_WECOM_AIBOT_REPLY_TIMEOUT"` + MaxSteps int `json:"max_steps" env:"PICOCLAW_CHANNELS_WECOM_AIBOT_MAX_STEPS"` // Maximum streaming steps +} + type PicoConfig struct { Enabled bool `json:"enabled" env:"PICOCLAW_CHANNELS_PICO_ENABLED"` Token string `json:"token" env:"PICOCLAW_CHANNELS_PICO_TOKEN"` diff --git a/pkg/config/defaults.go b/pkg/config/defaults.go index d19ce1d38..6b8118ebc 100644 --- a/pkg/config/defaults.go +++ b/pkg/config/defaults.go @@ -119,6 +119,17 @@ func DefaultConfig() *Config { AllowFrom: FlexibleStringSlice{}, ReplyTimeout: 5, }, + WeComAIBot: WeComAIBotConfig{ + Enabled: false, + Token: "", + EncodingAESKey: "", + WebhookHost: "0.0.0.0", + WebhookPort: 18791, + WebhookPath: "/webhook/wecom-aibot", + AllowFrom: FlexibleStringSlice{}, + ReplyTimeout: 5, + MaxSteps: 10, + }, Pico: PicoConfig{ Enabled: false, Token: "", From 1161aee8729bf44659dc70fc1060869b7c4c1208 Mon Sep 17 00:00:00 2001 From: mosir Date: Sat, 28 Feb 2026 00:18:10 +0800 Subject: [PATCH 5/7] fix(config): keep empty agents.defaults.model in saved config --- pkg/config/config.go | 2 +- pkg/config/config_test.go | 20 ++++++++++++++++++++ 2 files changed, 21 insertions(+), 1 deletion(-) diff --git a/pkg/config/config.go b/pkg/config/config.go index 2e0215278..d84772d2b 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -172,7 +172,7 @@ type AgentDefaults struct { RestrictToWorkspace bool `json:"restrict_to_workspace" env:"PICOCLAW_AGENTS_DEFAULTS_RESTRICT_TO_WORKSPACE"` Provider string `json:"provider" env:"PICOCLAW_AGENTS_DEFAULTS_PROVIDER"` ModelName string `json:"model_name,omitempty" env:"PICOCLAW_AGENTS_DEFAULTS_MODEL_NAME"` - Model string `json:"model,omitempty" env:"PICOCLAW_AGENTS_DEFAULTS_MODEL"` // Deprecated: use model_name instead + Model string `json:"model" env:"PICOCLAW_AGENTS_DEFAULTS_MODEL"` // Deprecated: use model_name instead ModelFallbacks []string `json:"model_fallbacks,omitempty"` ImageModel string `json:"image_model,omitempty" env:"PICOCLAW_AGENTS_DEFAULTS_IMAGE_MODEL"` ImageModelFallbacks []string `json:"image_model_fallbacks,omitempty"` diff --git a/pkg/config/config_test.go b/pkg/config/config_test.go index 45cdd8ec8..12fd10b50 100644 --- a/pkg/config/config_test.go +++ b/pkg/config/config_test.go @@ -5,6 +5,7 @@ import ( "os" "path/filepath" "runtime" + "strings" "testing" ) @@ -324,6 +325,25 @@ func TestSaveConfig_FilePermissions(t *testing.T) { } } +func TestSaveConfig_IncludesEmptyLegacyModelField(t *testing.T) { + tmpDir := t.TempDir() + path := filepath.Join(tmpDir, "config.json") + + cfg := DefaultConfig() + if err := SaveConfig(path, cfg); err != nil { + t.Fatalf("SaveConfig failed: %v", err) + } + + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("ReadFile failed: %v", err) + } + + if !strings.Contains(string(data), `"model": ""`) { + t.Fatalf("saved config should include empty legacy model field, got: %s", string(data)) + } +} + // TestConfig_Complete verifies all config fields are set func TestConfig_Complete(t *testing.T) { cfg := DefaultConfig() From ebfa72a286cc83ebf706fb2b90ffd4f64a2d3475 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=BE=8E=E9=9B=BB=E7=90=83?= Date: Sat, 28 Feb 2026 00:36:39 +0800 Subject: [PATCH 6/7] Update config file reference from config.yaml to config.json Closes: #565 --- pkg/channels/telegram/telegram_commands.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/channels/telegram/telegram_commands.go b/pkg/channels/telegram/telegram_commands.go index ee3bfef51..496fc5e4f 100644 --- a/pkg/channels/telegram/telegram_commands.go +++ b/pkg/channels/telegram/telegram_commands.go @@ -119,7 +119,7 @@ func (c *cmd) List(ctx context.Context, message telego.Message) error { if provider == "" { provider = "configured default" } - response = fmt.Sprintf("Configured Model: %s\nProvider: %s\n\nTo change models, update config.yaml", + response = fmt.Sprintf("Configured Model: %s\nProvider: %s\n\nTo change models, update config.json", c.config.Agents.Defaults.GetModelName(), provider) case "channels": From 43014cd36eeb34255d35ca3eb158e35dac7cd83d Mon Sep 17 00:00:00 2001 From: Zhang Rui Date: Sat, 28 Feb 2026 12:47:14 +0800 Subject: [PATCH 7/7] fix(channel): Add welcome message configuration and update related processing logic --- config/config.example.json | 3 +- pkg/channels/wecom/aibot.go | 72 +++++++++++++++++++++++++++++++------ pkg/config/config.go | 3 +- pkg/config/defaults.go | 1 + 4 files changed, 67 insertions(+), 12 deletions(-) diff --git a/config/config.example.json b/config/config.example.json index 8a8077cd0..36fb6ce71 100644 --- a/config/config.example.json +++ b/config/config.example.json @@ -162,7 +162,8 @@ "webhook_host": "0.0.0.0", "webhook_port": 18791, "webhook_path": "/webhook/wecom-aibot", - "max_steps": 10 + "max_steps": 10, + "welcome_message": "Hello! I'm your AI assistant. How can I help you today?" } }, "providers": { diff --git a/pkg/channels/wecom/aibot.go b/pkg/channels/wecom/aibot.go index d78f130bd..459071cd2 100644 --- a/pkg/channels/wecom/aibot.go +++ b/pkg/channels/wecom/aibot.go @@ -89,7 +89,9 @@ type WeComAIBotMessage struct { } `json:"msg_item"` } `json:"mixed,omitempty"` // event field - Event string `json:"event,omitempty"` + Event *struct { + EventType string `json:"eventtype"` + } `json:"event,omitempty"` } // WeComAIBotStreamResponse represents the streaming response format @@ -423,7 +425,7 @@ func (c *WeComAIBotChannel) processMessage(ctx context.Context, msg WeComAIBotMe }{ ID: c.generateStreamID(), Finish: true, - Content: "暂不支持该消息类型", + Content: "Unsupported message type: " + msg.MsgType, }, }) } @@ -545,6 +547,7 @@ func (c *WeComAIBotChannel) handleStreamMessage(ctx context.Context, msg WeComAI // handleImageMessage handles image messages func (c *WeComAIBotChannel) handleImageMessage(ctx context.Context, msg WeComAIBotMessage, timestamp, nonce string) string { + logger.WarnC("wecom_aibot", "Image message type not yet fully implemented") if msg.Image == nil { logger.ErrorC("wecom_aibot", "Image message missing image field") return c.encryptEmptyResponse(timestamp, nonce) @@ -553,7 +556,7 @@ func (c *WeComAIBotChannel) handleImageMessage(ctx context.Context, msg WeComAIB imageURL := msg.Image.URL // Download and decrypt image - imageData, err := c.downloadAndDecryptImage(ctx, imageURL) + _, err := c.downloadAndDecryptImage(ctx, imageURL) if err != nil { logger.ErrorCF("wecom_aibot", "Failed to process image", map[string]any{ "error": err, @@ -575,14 +578,35 @@ func (c *WeComAIBotChannel) handleImageMessage(ctx context.Context, msg WeComAIB }{ ID: c.generateStreamID(), Finish: true, - Content: fmt.Sprintf("图片处理失败: %v", err), + Content: fmt.Sprintf("Image received (URL: %s), but image messages are not yet supported", imageURL), }, }) } // Echo back the image (simple demo behavior) - streamID := c.generateStreamID() - return c.encryptImageResponse(streamID, timestamp, nonce, imageData) + // streamID := c.generateStreamID() + // return c.encryptImageResponse(streamID, timestamp, nonce, imageData) + + // For now, just acknowledge receipt without echoing the image + return c.encryptResponse("", timestamp, nonce, WeComAIBotStreamResponse{ + MsgType: "stream", + Stream: struct { + ID string `json:"id"` + Finish bool `json:"finish"` + Content string `json:"content,omitempty"` + MsgItem []struct { + MsgType string `json:"msgtype"` + Image *struct { + Base64 string `json:"base64"` + MD5 string `json:"md5"` + } `json:"image,omitempty"` + } `json:"msg_item,omitempty"` + }{ + ID: c.generateStreamID(), + Finish: true, + Content: fmt.Sprintf("Image received (URL: %s), but image messages are not yet supported", imageURL), + }, + }) } // handleMixedMessage handles mixed (text + image) messages @@ -604,17 +628,45 @@ func (c *WeComAIBotChannel) handleMixedMessage(ctx context.Context, msg WeComAIB }{ ID: c.generateStreamID(), Finish: true, - Content: "暂不支持图文混排消息", + Content: "Mixed message type is not yet supported", }, }) } // handleEventMessage handles event messages func (c *WeComAIBotChannel) handleEventMessage(ctx context.Context, msg WeComAIBotMessage, timestamp, nonce string) string { + eventType := "" + if msg.Event != nil { + eventType = msg.Event.EventType + } logger.DebugCF("wecom_aibot", "Received event", map[string]any{ - "event": msg.Event, + "event_type": eventType, }) - // Return empty response for events + + // Send welcome message when user opens the chat window + if eventType == "enter_chat" && c.config.WelcomeMessage != "" { + streamID := c.generateStreamID() + return c.encryptResponse(streamID, timestamp, nonce, WeComAIBotStreamResponse{ + MsgType: "stream", + Stream: struct { + ID string `json:"id"` + Finish bool `json:"finish"` + Content string `json:"content,omitempty"` + MsgItem []struct { + MsgType string `json:"msgtype"` + Image *struct { + Base64 string `json:"base64"` + MD5 string `json:"md5"` + } `json:"image,omitempty"` + } `json:"msg_item,omitempty"` + }{ + ID: streamID, + Finish: true, + Content: c.config.WelcomeMessage, + }, + }) + } + return c.encryptEmptyResponse(timestamp, nonce) } @@ -862,7 +914,7 @@ func (c *WeComAIBotChannel) encryptMessage(plaintext, receiveid string) (string, // Generate 16-byte random string randomBytes := make([]byte, 16) - for i := 0; i < 16; i++ { + for i := range 16 { n, err := rand.Int(rand.Reader, big.NewInt(10)) if err != nil { return "", fmt.Errorf("failed to generate random: %w", err) diff --git a/pkg/config/config.go b/pkg/config/config.go index c6bfe8404..d8a0b52a5 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -369,7 +369,8 @@ type WeComAIBotConfig struct { WebhookPath string `json:"webhook_path" env:"PICOCLAW_CHANNELS_WECOM_AIBOT_WEBHOOK_PATH"` AllowFrom FlexibleStringSlice `json:"allow_from" env:"PICOCLAW_CHANNELS_WECOM_AIBOT_ALLOW_FROM"` ReplyTimeout int `json:"reply_timeout" env:"PICOCLAW_CHANNELS_WECOM_AIBOT_REPLY_TIMEOUT"` - MaxSteps int `json:"max_steps" env:"PICOCLAW_CHANNELS_WECOM_AIBOT_MAX_STEPS"` // Maximum streaming steps + MaxSteps int `json:"max_steps" env:"PICOCLAW_CHANNELS_WECOM_AIBOT_MAX_STEPS"` // Maximum streaming steps + WelcomeMessage string `json:"welcome_message" env:"PICOCLAW_CHANNELS_WECOM_AIBOT_WELCOME_MESSAGE"` // Sent on enter_chat event; empty = no welcome } type PicoConfig struct { diff --git a/pkg/config/defaults.go b/pkg/config/defaults.go index 3047f0536..abe8bbc7c 100644 --- a/pkg/config/defaults.go +++ b/pkg/config/defaults.go @@ -131,6 +131,7 @@ func DefaultConfig() *Config { AllowFrom: FlexibleStringSlice{}, ReplyTimeout: 5, MaxSteps: 10, + WelcomeMessage: "Hello! I'm your AI assistant. How can I help you today?", }, Pico: PicoConfig{ Enabled: false,