From 6a7bf32f9e1e6856ccf1028a2dc6167eb16a7f85 Mon Sep 17 00:00:00 2001 From: Cytown Date: Tue, 28 Apr 2026 12:13:06 +0800 Subject: [PATCH] refactor: standardize channel identification and decouple name from provider type Decouple channel names (config keys) from channel types (provider IDs) to allow multiple instances of the same provider. This ensures robust identification across the message bus and agent dispatch logic. - Introduce ChannelType in InboundContext for consistent provider tracking - Refactor IsInternalChannel to use provider type instead of generic names - Update all channel implementations to pass through their provider type via BaseChannel - Standardize agent dispatch and heartbeat recording to use ChannelType - Enhance shell and cron tools with improved execution tracking and fallback logic - Fix redundant channel field assignments and hardcoded types in providers - Resolve build issues by adding missing logger imports to tool packages --- pkg/agent/agent.go | 2 +- pkg/agent/agent_message.go | 20 ++++++------ pkg/agent/agent_test.go | 6 ++-- pkg/agent/agent_utils.go | 3 ++ pkg/agent/dispatch_request.go | 11 +++++++ pkg/agent/pipeline_llm.go | 11 ++++--- pkg/audio/asr/agent.go | 9 +++--- pkg/bus/inbound_context.go | 6 ++++ pkg/bus/types.go | 5 +-- pkg/channels/base.go | 12 +++++++ pkg/channels/dingtalk/dingtalk.go | 1 + pkg/channels/discord/discord.go | 2 +- pkg/channels/feishu/feishu_64.go | 1 + pkg/channels/irc/irc.go | 1 + pkg/channels/line/line.go | 1 + pkg/channels/maixcam/maixcam.go | 1 + pkg/channels/manager.go | 14 ++++++++- pkg/channels/matrix/matrix.go | 1 + pkg/channels/onebot/onebot.go | 1 + pkg/channels/pico/client.go | 4 ++- pkg/channels/pico/pico.go | 17 +++++----- pkg/channels/qq/qq.go | 1 + pkg/channels/slack/slack.go | 4 +-- pkg/channels/teams_webhook/teams_webhook.go | 1 + pkg/channels/telegram/telegram.go | 2 +- pkg/channels/vk/vk.go | 1 + pkg/channels/wecom/wecom.go | 1 + pkg/channels/weixin/weixin.go | 1 + pkg/channels/whatsapp/whatsapp.go | 1 + .../whatsapp_native/whatsapp_native.go | 5 ++- pkg/constants/channels.go | 4 +-- pkg/tools/cron.go | 20 +++++++++++- pkg/tools/shell.go | 31 ++++++++++++++++--- 33 files changed, 156 insertions(+), 45 deletions(-) diff --git a/pkg/agent/agent.go b/pkg/agent/agent.go index 2c456dca7..20abd74b6 100644 --- a/pkg/agent/agent.go +++ b/pkg/agent/agent.go @@ -474,7 +474,7 @@ func (al *AgentLoop) runAgentLoop( // Record last channel for heartbeat notifications (skip internal channels and cli) if opts.Dispatch.Channel() != "" && opts.Dispatch.ChatID() != "" && - !constants.IsInternalChannel(opts.Dispatch.Channel()) { + !constants.IsInternalChannel(opts.Dispatch.ChannelType()) { channelKey := fmt.Sprintf("%s:%s", opts.Dispatch.Channel(), opts.Dispatch.ChatID()) if err := al.RecordLastChannel(channelKey); err != nil { logger.WarnCF( diff --git a/pkg/agent/agent_message.go b/pkg/agent/agent_message.go index 96b0b0817..714a9c33a 100644 --- a/pkg/agent/agent_message.go +++ b/pkg/agent/agent_message.go @@ -53,10 +53,11 @@ func (al *AgentLoop) ProcessDirectWithChannel( msg := bus.InboundMessage{ Context: bus.InboundContext{ - Channel: channel, - ChatID: chatID, - ChatType: "direct", - SenderID: "cron", + Channel: channel, + ChannelType: channel, // For direct calls, channel name equals channel type + ChatID: chatID, + ChatType: "direct", + SenderID: "cron", }, Content: content, SessionKey: sessionKey, @@ -86,10 +87,11 @@ func (al *AgentLoop) ProcessHeartbeat( } if channel != "" || chatID != "" { dispatch.InboundContext = &bus.InboundContext{ - Channel: channel, - ChatID: chatID, - ChatType: "direct", - SenderID: "heartbeat", + Channel: channel, + ChannelType: channel, // For heartbeat, channel name equals channel type + ChatID: chatID, + ChatType: "direct", + SenderID: "heartbeat", } } return al.runAgentLoop(ctx, agent, processOptions{ @@ -247,7 +249,7 @@ func (al *AgentLoop) processSystemMessage( // Parse origin channel from chat_id (format: "channel:chat_id") var originChannel, originChatID string if idx := strings.Index(msg.ChatID, ":"); idx > 0 { - originChannel = msg.ChatID[:idx] + originChannel = msg.ChatID[:idx] // e.g. "telegram" originChatID = msg.ChatID[idx+1:] } else { originChannel = "cli" diff --git a/pkg/agent/agent_test.go b/pkg/agent/agent_test.go index 4047ab74d..6ec8db913 100644 --- a/pkg/agent/agent_test.go +++ b/pkg/agent/agent_test.go @@ -3911,7 +3911,8 @@ func TestProcessMessage_PicoPublishesReasoningAsThoughtMessage(t *testing.T) { al := NewAgentLoop(cfg, msgBus, provider) response, err := al.processMessage(context.Background(), bus.InboundMessage{ - Channel: "pico", + Context: bus.InboundContext{Channel: "pico1", ChannelType: "pico"}, + Channel: "pico1", SenderID: "user1", ChatID: "pico:test-session", Content: "hello", @@ -4471,7 +4472,8 @@ func TestRun_PicoPublishesAssistantContentDuringToolCallsWithoutFinalDuplicate(t }() if err := msgBus.PublishInbound(context.Background(), bus.InboundMessage{ - Channel: "pico", + Context: bus.InboundContext{Channel: "pico1", ChannelType: "pico"}, + Channel: "pico1", SenderID: "user-1", ChatID: "session-1", Content: "run with tools", diff --git a/pkg/agent/agent_utils.go b/pkg/agent/agent_utils.go index bbfb3f2ae..238b5cf39 100644 --- a/pkg/agent/agent_utils.go +++ b/pkg/agent/agent_utils.go @@ -217,6 +217,9 @@ func appendEventContextFields(fields map[string]any, turnCtx *TurnContext) { if inbound.Channel != "" { fields["inbound_channel"] = inbound.Channel } + if inbound.ChannelType != "" { + fields["inbound_channel_type"] = inbound.ChannelType + } if inbound.Account != "" { fields["inbound_account"] = inbound.Account } diff --git a/pkg/agent/dispatch_request.go b/pkg/agent/dispatch_request.go index cb54264d6..a3693f127 100644 --- a/pkg/agent/dispatch_request.go +++ b/pkg/agent/dispatch_request.go @@ -27,6 +27,13 @@ func (r DispatchRequest) Channel() string { return r.InboundContext.Channel } +func (r DispatchRequest) ChannelType() string { + if r.InboundContext == nil { + return "" + } + return r.InboundContext.ChannelType +} + func (r DispatchRequest) ChatID() string { if r.InboundContext == nil { return "" @@ -93,6 +100,10 @@ func normalizeProcessOptions(opts processOptions) processOptions { MessageID: strings.TrimSpace(opts.MessageID), ReplyToMessageID: strings.TrimSpace(opts.ReplyToMessageID), } + // Set ChannelType from Channel if not already set + if inbound.ChannelType == "" && inbound.Channel != "" { + inbound.ChannelType = inbound.Channel + } inbound.ChatType = inferChatTypeFromSessionScope(opts.Dispatch.SessionScope) if inbound.Channel != "" || inbound.ChatID != "" || inbound.SenderID != "" || inbound.MessageID != "" || inbound.ReplyToMessageID != "" { diff --git a/pkg/agent/pipeline_llm.go b/pkg/agent/pipeline_llm.go index 6bf55fa39..73c9eb2e8 100644 --- a/pkg/agent/pipeline_llm.go +++ b/pkg/agent/pipeline_llm.go @@ -10,6 +10,7 @@ import ( "strings" "time" + "github.com/sipeed/picoclaw/pkg/config" "github.com/sipeed/picoclaw/pkg/constants" "github.com/sipeed/picoclaw/pkg/logger" "github.com/sipeed/picoclaw/pkg/providers" @@ -292,7 +293,7 @@ func (p *Pipeline) CallLLM( }, ) - if retry == 0 && !constants.IsInternalChannel(ts.channel) { + if retry == 0 && !constants.IsInternalChannel(ts.opts.Dispatch.ChannelType()) { al.bus.PublishOutbound(ctx, outboundMessageForTurn( ts, "Context window exceeded. Compressing history and retrying...", @@ -382,11 +383,12 @@ func (p *Pipeline) CallLLM( } reasoningContent := responseReasoningContent(exec.response) - shouldPublishPicoToolCallInterim := ts.channel == "pico" && len(exec.response.ToolCalls) > 0 + shouldPublishPicoToolCallInterim := ts.opts.Dispatch.ChannelType() == config.ChannelPico && + len(exec.response.ToolCalls) > 0 if shouldPublishPicoToolCallInterim { // Pico tool-call turns publish their reasoning/content/tool summary as a // structured sequence after the tool-call payload is normalized below. - } else if ts.channel == "pico" { + } else if ts.opts.Dispatch.ChannelType() == config.ChannelPico { go al.publishPicoReasoning(turnCtx, reasoningContent, ts.chatID) } else { go al.handleReasoning( @@ -425,7 +427,8 @@ func (p *Pipeline) CallLLM( // No-tool-call path: steering check and direct response if len(exec.response.ToolCalls) == 0 || exec.gracefulTerminal { responseContent := exec.response.Content - if responseContent == "" && exec.response.ReasoningContent != "" && ts.channel != "pico" { + if responseContent == "" && exec.response.ReasoningContent != "" && + ts.opts.Dispatch.ChannelType() != config.ChannelPico { responseContent = exec.response.ReasoningContent } if steerMsgs := al.dequeueSteeringMessagesForScope(ts.sessionKey); len(steerMsgs) > 0 { diff --git a/pkg/audio/asr/agent.go b/pkg/audio/asr/agent.go index c483a0778..4e389743d 100644 --- a/pkg/audio/asr/agent.go +++ b/pkg/audio/asr/agent.go @@ -238,10 +238,11 @@ func (a *Agent) processUtterance(ctx context.Context, acc *speechAccumulator) { if err := a.bus.PublishInbound(ctx, bus.InboundMessage{ Context: bus.InboundContext{ - Channel: channelType, - ChatID: acc.chatID, - ChatType: "channel", - SenderID: acc.speakerID, + Channel: channelType, + ChannelType: channelType, + ChatID: acc.chatID, + ChatType: "channel", + SenderID: acc.speakerID, Raw: map[string]string{ "is_voice": "true", }, diff --git a/pkg/bus/inbound_context.go b/pkg/bus/inbound_context.go index d6be80565..bd7910cac 100644 --- a/pkg/bus/inbound_context.go +++ b/pkg/bus/inbound_context.go @@ -32,6 +32,7 @@ func NormalizeInboundMessage(msg InboundMessage) InboundMessage { func (ctx InboundContext) isZero() bool { return ctx.Channel == "" && + ctx.ChannelType == "" && ctx.Account == "" && ctx.ChatID == "" && ctx.ChatType == "" && @@ -49,6 +50,11 @@ func (ctx InboundContext) isZero() bool { func normalizeInboundContext(ctx InboundContext) InboundContext { ctx.Channel = strings.TrimSpace(ctx.Channel) + ctx.ChannelType = strings.TrimSpace(ctx.ChannelType) + // Set ChannelType from Channel if not already set + if ctx.ChannelType == "" && ctx.Channel != "" { + ctx.ChannelType = ctx.Channel + } ctx.Account = strings.TrimSpace(ctx.Account) ctx.ChatID = strings.TrimSpace(ctx.ChatID) ctx.ChatType = normalizeKind(ctx.ChatType) diff --git a/pkg/bus/types.go b/pkg/bus/types.go index 953e69d9c..95754f7af 100644 --- a/pkg/bus/types.go +++ b/pkg/bus/types.go @@ -13,8 +13,9 @@ type SenderInfo struct { // inbound message. This is the source of truth for routing and session // allocation. type InboundContext struct { - Channel string `json:"channel"` - Account string `json:"account,omitempty"` + Channel string `json:"channel"` + ChannelType string `json:"channel_type,omitempty"` // telegram, discord, slack, etc. + Account string `json:"account,omitempty"` ChatID string `json:"chat_id"` ChatType string `json:"chat_type,omitempty"` // direct / group / channel diff --git a/pkg/channels/base.go b/pkg/channels/base.go index 3585fb075..19bf30dc3 100644 --- a/pkg/channels/base.go +++ b/pkg/channels/base.go @@ -75,6 +75,11 @@ func WithReasoningChannelID(id string) BaseChannelOption { return func(c *BaseChannel) { c.reasoningChannelID = id } } +// WithChannelType sets the channel type (from config.Channel.Type). +func WithChannelType(channelType string) BaseChannelOption { + return func(c *BaseChannel) { c.channelType = channelType } +} + // MessageLengthProvider is an opt-in interface that channels implement // to advertise their maximum message length. The Manager uses this via // type assertion to decide whether to split outbound messages. @@ -87,6 +92,7 @@ type BaseChannel struct { bus *bus.MessageBus running atomic.Bool name string + channelType string allowList []string maxMessageLength int groupTrigger config.GroupTriggerConfig @@ -187,6 +193,11 @@ func (c *BaseChannel) Name() string { return c.name } +// ChannelType returns the channel type (from config.Channel.Type). +func (c *BaseChannel) ChannelType() string { + return c.channelType +} + // SetName updates the channel name. Used by the manager after channel creation // to ensure the name matches the config key (which may differ from the type). func (c *BaseChannel) SetName(name string) { @@ -294,6 +305,7 @@ func (c *BaseChannel) HandleMessageWithContext( } inboundCtx.Channel = c.name + inboundCtx.ChannelType = c.channelType if inboundCtx.ChatID == "" { inboundCtx.ChatID = deliveryChatID } diff --git a/pkg/channels/dingtalk/dingtalk.go b/pkg/channels/dingtalk/dingtalk.go index 9cd461bc8..581be6d1f 100644 --- a/pkg/channels/dingtalk/dingtalk.go +++ b/pkg/channels/dingtalk/dingtalk.go @@ -52,6 +52,7 @@ func NewDingTalkChannel( channels.WithMaxMessageLength(20000), channels.WithGroupTrigger(bc.GroupTrigger), channels.WithReasoningChannelID(bc.ReasoningChannelID), + channels.WithChannelType(bc.Type), ) return &DingTalkChannel{ diff --git a/pkg/channels/discord/discord.go b/pkg/channels/discord/discord.go index 514b9b3b1..4f0003e9d 100644 --- a/pkg/channels/discord/discord.go +++ b/pkg/channels/discord/discord.go @@ -85,6 +85,7 @@ func NewDiscordChannel( channels.WithMaxMessageLength(2000), channels.WithGroupTrigger(bc.GroupTrigger), channels.WithReasoningChannelID(bc.ReasoningChannelID), + channels.WithChannelType(bc.Type), ) ch := &DiscordChannel{ @@ -669,7 +670,6 @@ func (c *DiscordChannel) handleMessage(s *discordgo.Session, m *discordgo.Messag "is_dm": fmt.Sprintf("%t", m.GuildID == ""), } inboundCtx := bus.InboundContext{ - Channel: c.Name(), ChatID: m.ChannelID, ChatType: peerKind, SenderID: senderID, diff --git a/pkg/channels/feishu/feishu_64.go b/pkg/channels/feishu/feishu_64.go index 8f3ae39d9..d411c6114 100644 --- a/pkg/channels/feishu/feishu_64.go +++ b/pkg/channels/feishu/feishu_64.go @@ -63,6 +63,7 @@ func NewFeishuChannel(bc *config.Channel, cfg *config.FeishuSettings, bus *bus.M base := channels.NewBaseChannel("feishu", cfg, bus, bc.AllowFrom, channels.WithGroupTrigger(bc.GroupTrigger), channels.WithReasoningChannelID(bc.ReasoningChannelID), + channels.WithChannelType(bc.Type), ) tc := newTokenCache() diff --git a/pkg/channels/irc/irc.go b/pkg/channels/irc/irc.go index fa60e9b6d..a56524b93 100644 --- a/pkg/channels/irc/irc.go +++ b/pkg/channels/irc/irc.go @@ -38,6 +38,7 @@ func NewIRCChannel(bc *config.Channel, cfg *config.IRCSettings, messageBus *bus. channels.WithMaxMessageLength(400), channels.WithGroupTrigger(bc.GroupTrigger), channels.WithReasoningChannelID(bc.ReasoningChannelID), + channels.WithChannelType(bc.Type), ) return &IRCChannel{ diff --git a/pkg/channels/line/line.go b/pkg/channels/line/line.go index 760506a31..5e2c19166 100644 --- a/pkg/channels/line/line.go +++ b/pkg/channels/line/line.go @@ -74,6 +74,7 @@ func NewLINEChannel( channels.WithMaxMessageLength(5000), channels.WithGroupTrigger(bc.GroupTrigger), channels.WithReasoningChannelID(bc.ReasoningChannelID), + channels.WithChannelType(bc.Type), ) return &LINEChannel{ diff --git a/pkg/channels/maixcam/maixcam.go b/pkg/channels/maixcam/maixcam.go index b81206c59..62cd7fe90 100644 --- a/pkg/channels/maixcam/maixcam.go +++ b/pkg/channels/maixcam/maixcam.go @@ -43,6 +43,7 @@ func NewMaixCamChannel( bus, bc.AllowFrom, channels.WithReasoningChannelID(bc.ReasoningChannelID), + channels.WithChannelType(bc.Type), ) return &MaixCamChannel{ diff --git a/pkg/channels/manager.go b/pkg/channels/manager.go index d56c4fd9b..4aaaf146d 100644 --- a/pkg/channels/manager.go +++ b/pkg/channels/manager.go @@ -123,6 +123,10 @@ func outboundMessageChannel(msg bus.OutboundMessage) string { return msg.Context.Channel } +func outboundMessageChannelType(msg bus.OutboundMessage) string { + return msg.Context.ChannelType +} + func outboundMessageChatID(msg bus.OutboundMessage) string { return msg.ChatID } @@ -146,6 +150,10 @@ func outboundMediaChannel(msg bus.OutboundMediaMessage) string { return msg.Context.Channel } +func outboundMediaChannelType(msg bus.OutboundMediaMessage) string { + return msg.Context.ChannelType +} + func outboundMediaChatID(msg bus.OutboundMediaMessage) string { return msg.ChatID } @@ -1066,6 +1074,7 @@ func dispatchLoop[M any]( m *Manager, ch <-chan M, getChannel func(M) string, + getChannelType func(M) string, enqueue func(context.Context, *channelWorker, M) bool, startMsg, stopMsg, unknownMsg, noWorkerMsg string, ) { @@ -1084,9 +1093,10 @@ func dispatchLoop[M any]( } channel := getChannel(msg) + channelType := getChannelType(msg) // Silently skip internal channels - if constants.IsInternalChannel(channel) { + if constants.IsInternalChannel(channelType) { continue } @@ -1116,6 +1126,7 @@ func (m *Manager) dispatchOutbound(ctx context.Context) { ctx, m, m.bus.OutboundChan(), func(msg bus.OutboundMessage) string { return outboundMessageChannel(msg) }, + func(msg bus.OutboundMessage) string { return outboundMessageChannelType(msg) }, func(ctx context.Context, w *channelWorker, msg bus.OutboundMessage) bool { select { case w.queue <- msg: @@ -1136,6 +1147,7 @@ func (m *Manager) dispatchOutboundMedia(ctx context.Context) { ctx, m, m.bus.OutboundMediaChan(), func(msg bus.OutboundMediaMessage) string { return outboundMediaChannel(msg) }, + func(msg bus.OutboundMediaMessage) string { return outboundMediaChannelType(msg) }, func(ctx context.Context, w *channelWorker, msg bus.OutboundMediaMessage) bool { select { case w.mediaQueue <- msg: diff --git a/pkg/channels/matrix/matrix.go b/pkg/channels/matrix/matrix.go index 04599d6d2..d41ce5ace 100644 --- a/pkg/channels/matrix/matrix.go +++ b/pkg/channels/matrix/matrix.go @@ -242,6 +242,7 @@ func NewMatrixChannel( channels.WithMaxMessageLength(65536), channels.WithGroupTrigger(bc.GroupTrigger), channels.WithReasoningChannelID(bc.ReasoningChannelID), + channels.WithChannelType(bc.Type), ) ch := &MatrixChannel{ diff --git a/pkg/channels/onebot/onebot.go b/pkg/channels/onebot/onebot.go index f0d0a890f..8fab76142 100644 --- a/pkg/channels/onebot/onebot.go +++ b/pkg/channels/onebot/onebot.go @@ -104,6 +104,7 @@ func NewOneBotChannel( base := channels.NewBaseChannel("onebot", cfg, messageBus, bc.AllowFrom, channels.WithGroupTrigger(bc.GroupTrigger), channels.WithReasoningChannelID(bc.ReasoningChannelID), + channels.WithChannelType(bc.Type), ) const dedupSize = 1024 diff --git a/pkg/channels/pico/client.go b/pkg/channels/pico/client.go index 009900e01..0dac0facb 100644 --- a/pkg/channels/pico/client.go +++ b/pkg/channels/pico/client.go @@ -39,7 +39,9 @@ func NewPicoClientChannel( return nil, fmt.Errorf("pico_client url is required") } - base := channels.NewBaseChannel("pico_client", cfg, messageBus, bc.AllowFrom) + base := channels.NewBaseChannel("pico_client", cfg, messageBus, bc.AllowFrom, + channels.WithChannelType(bc.Type), + ) return &PicoClientChannel{ BaseChannel: base, diff --git a/pkg/channels/pico/pico.go b/pkg/channels/pico/pico.go index d1de8f4d5..1c1e120b8 100644 --- a/pkg/channels/pico/pico.go +++ b/pkg/channels/pico/pico.go @@ -117,7 +117,9 @@ func NewPicoChannel( return nil, fmt.Errorf("pico token is required") } - base := channels.NewBaseChannel("pico", cfg, messageBus, bc.AllowFrom) + base := channels.NewBaseChannel("pico", cfg, messageBus, bc.AllowFrom, + channels.WithChannelType(bc.Type), + ) allowOrigins := cfg.AllowOrigins checkOrigin := func(r *http.Request) bool { @@ -965,12 +967,13 @@ func (c *PicoChannel) handleMessageSend(pc *picoConn, msg PicoMessage) { } inboundCtx := bus.InboundContext{ - Channel: "pico", - ChatID: chatID, - ChatType: "direct", - SenderID: senderID, - MessageID: msg.ID, - Raw: metadata, + Channel: c.bc.Name(), + ChannelType: config.ChannelPico, + ChatID: chatID, + ChatType: "direct", + SenderID: senderID, + MessageID: msg.ID, + Raw: metadata, } c.HandleInboundContext(c.ctx, chatID, content, media, inboundCtx, sender) diff --git a/pkg/channels/qq/qq.go b/pkg/channels/qq/qq.go index 71cba5548..0ce5244fe 100644 --- a/pkg/channels/qq/qq.go +++ b/pkg/channels/qq/qq.go @@ -88,6 +88,7 @@ func NewQQChannel(bc *config.Channel, cfg *config.QQSettings, messageBus *bus.Me channels.WithMaxMessageLength(cfg.MaxMessageLength), channels.WithGroupTrigger(bc.GroupTrigger), channels.WithReasoningChannelID(bc.ReasoningChannelID), + channels.WithChannelType(bc.Type), ) return &QQChannel{ diff --git a/pkg/channels/slack/slack.go b/pkg/channels/slack/slack.go index 19e7b737c..adc22e91f 100644 --- a/pkg/channels/slack/slack.go +++ b/pkg/channels/slack/slack.go @@ -56,6 +56,7 @@ func NewSlackChannel( channels.WithMaxMessageLength(40000), channels.WithGroupTrigger(bc.GroupTrigger), channels.WithReasoningChannelID(bc.ReasoningChannelID), + channels.WithChannelType(bc.Type), ) return &SlackChannel{ @@ -381,7 +382,6 @@ func (c *SlackChannel) handleMessageEvent(ev *slackevents.MessageEvent) { }) inboundCtx := bus.InboundContext{ - Channel: c.Name(), Account: c.teamID, ChatID: channelID, ChatType: peerKind, @@ -456,7 +456,6 @@ func (c *SlackChannel) handleAppMention(ev *slackevents.AppMentionEvent) { "team_id": c.teamID, } inboundCtx := bus.InboundContext{ - Channel: c.Name(), Account: c.teamID, ChatID: channelID, ChatType: mentionPeerKind, @@ -521,7 +520,6 @@ func (c *SlackChannel) handleSlashCommand(event socketmode.Event) { peerKind = "direct" } inboundCtx := bus.InboundContext{ - Channel: c.Name(), Account: c.teamID, ChatID: channelID, ChatType: peerKind, diff --git a/pkg/channels/teams_webhook/teams_webhook.go b/pkg/channels/teams_webhook/teams_webhook.go index 837563453..1a526b737 100644 --- a/pkg/channels/teams_webhook/teams_webhook.go +++ b/pkg/channels/teams_webhook/teams_webhook.go @@ -95,6 +95,7 @@ func NewTeamsWebhookChannel( "*", }, // Output-only channel; "*" suppresses misleading "allows EVERYONE" audit warning channels.WithMaxMessageLength(24000), // Power Automate webhook payload limit is 28KB + channels.WithChannelType(bc.Type), ) client := goteamsnotify.NewTeamsClient() diff --git a/pkg/channels/telegram/telegram.go b/pkg/channels/telegram/telegram.go index cebebfed6..9a6972885 100644 --- a/pkg/channels/telegram/telegram.go +++ b/pkg/channels/telegram/telegram.go @@ -104,6 +104,7 @@ func NewTelegramChannel( channels.WithMaxMessageLength(4000), channels.WithGroupTrigger(bc.GroupTrigger), channels.WithReasoningChannelID(bc.ReasoningChannelID), + channels.WithChannelType(bc.Type), ) ch := &TelegramChannel{ @@ -891,7 +892,6 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes } inboundCtx := bus.InboundContext{ - Channel: c.Name(), ChatID: fmt.Sprintf("%d", chatID), ChatType: peerKind, SenderID: platformID, diff --git a/pkg/channels/vk/vk.go b/pkg/channels/vk/vk.go index b27431ba0..6341177f9 100644 --- a/pkg/channels/vk/vk.go +++ b/pkg/channels/vk/vk.go @@ -45,6 +45,7 @@ func NewVKChannel(channelName string, bc *config.Channel, bus *bus.MessageBus) ( channels.WithMaxMessageLength(4000), channels.WithGroupTrigger(bc.GroupTrigger), channels.WithReasoningChannelID(bc.ReasoningChannelID), + channels.WithChannelType(bc.Type), ) return &VKChannel{ diff --git a/pkg/channels/wecom/wecom.go b/pkg/channels/wecom/wecom.go index a0a23feda..7599d703f 100644 --- a/pkg/channels/wecom/wecom.go +++ b/pkg/channels/wecom/wecom.go @@ -122,6 +122,7 @@ func NewChannel(bc *config.Channel, cfg *config.WeComSettings, messageBus *bus.M messageBus, bc.AllowFrom, channels.WithReasoningChannelID(bc.ReasoningChannelID), + channels.WithChannelType(bc.Type), ) ch := &WeComChannel{ diff --git a/pkg/channels/weixin/weixin.go b/pkg/channels/weixin/weixin.go index 2897d2422..f0ecb9554 100644 --- a/pkg/channels/weixin/weixin.go +++ b/pkg/channels/weixin/weixin.go @@ -78,6 +78,7 @@ func NewWeixinChannel( bc.AllowFrom, channels.WithMaxMessageLength(4000), channels.WithReasoningChannelID(bc.ReasoningChannelID), + channels.WithChannelType(bc.Type), ) return &WeixinChannel{ diff --git a/pkg/channels/whatsapp/whatsapp.go b/pkg/channels/whatsapp/whatsapp.go index 4c338b5f4..a81a8c7ad 100644 --- a/pkg/channels/whatsapp/whatsapp.go +++ b/pkg/channels/whatsapp/whatsapp.go @@ -40,6 +40,7 @@ func NewWhatsAppChannel( bc.AllowFrom, channels.WithMaxMessageLength(65536), channels.WithReasoningChannelID(bc.ReasoningChannelID), + channels.WithChannelType(bc.Type), ) return &WhatsAppChannel{ diff --git a/pkg/channels/whatsapp_native/whatsapp_native.go b/pkg/channels/whatsapp_native/whatsapp_native.go index de4ecfd44..09665a91c 100644 --- a/pkg/channels/whatsapp_native/whatsapp_native.go +++ b/pkg/channels/whatsapp_native/whatsapp_native.go @@ -70,7 +70,10 @@ func NewWhatsAppNativeChannel( bus *bus.MessageBus, storePath string, ) (channels.Channel, error) { - base := channels.NewBaseChannel(name, cfg, bus, bc.AllowFrom, channels.WithMaxMessageLength(65536)) + base := channels.NewBaseChannel(name, cfg, bus, bc.AllowFrom, + channels.WithMaxMessageLength(65536), + channels.WithChannelType(bc.Type), + ) if storePath == "" { storePath = "whatsapp" } diff --git a/pkg/constants/channels.go b/pkg/constants/channels.go index 0a46e6cd9..cb6db46b9 100644 --- a/pkg/constants/channels.go +++ b/pkg/constants/channels.go @@ -10,7 +10,7 @@ var internalChannels = map[string]struct{}{ } // IsInternalChannel returns true if the channel is an internal channel. -func IsInternalChannel(channel string) bool { - _, found := internalChannels[channel] +func IsInternalChannel(channelType string) bool { + _, found := internalChannels[channelType] return found } diff --git a/pkg/tools/cron.go b/pkg/tools/cron.go index f2e6561df..a128e34ae 100644 --- a/pkg/tools/cron.go +++ b/pkg/tools/cron.go @@ -12,6 +12,7 @@ import ( "github.com/sipeed/picoclaw/pkg/config" "github.com/sipeed/picoclaw/pkg/constants" "github.com/sipeed/picoclaw/pkg/cron" + "github.com/sipeed/picoclaw/pkg/logger" "github.com/sipeed/picoclaw/pkg/utils" ) @@ -25,6 +26,7 @@ type JobExecutor interface { // CronTool provides scheduling capabilities for the agent type CronTool struct { + cfg *config.Config cronService *cron.CronService executor JobExecutor msgBus *bus.MessageBus @@ -59,6 +61,7 @@ func NewCronTool( execTool.SetTimeout(execTimeout) } return &CronTool{ + cfg: config, cronService: cronService, executor: executor, msgBus: msgBus, @@ -201,7 +204,22 @@ func (t *CronTool) addJob(ctx context.Context, args map[string]any) *ToolResult if !t.execEnabled { return ErrorResult("command execution is disabled") } - if !constants.IsInternalChannel(channel) { + + var channelType string + if t.cfg != nil { + if ch := t.cfg.Channels.Get(channel); ch != nil { + channelType = ch.Type + } + } + // Fallback: if channelType is not determined from config, use channelName + if channelType == "" { + channelType = channel + logger.DebugCF("cron", "Channel type not found in config, falling back to name", map[string]any{ + "channel_name": channel, + }) + } + + if !constants.IsInternalChannel(channelType) { return ErrorResult("scheduling command execution is restricted to internal channels") } if !t.allowCommand && !commandConfirm { diff --git a/pkg/tools/shell.go b/pkg/tools/shell.go index a570ac9ec..d519c31a4 100644 --- a/pkg/tools/shell.go +++ b/pkg/tools/shell.go @@ -21,6 +21,7 @@ import ( "github.com/sipeed/picoclaw/pkg/config" "github.com/sipeed/picoclaw/pkg/constants" "github.com/sipeed/picoclaw/pkg/isolation" + "github.com/sipeed/picoclaw/pkg/logger" ) var ( @@ -35,6 +36,7 @@ func getSessionManager() *SessionManager { } type ExecTool struct { + cfg *config.Config workingDir string timeout time.Duration denyPatterns []*regexp.Regexp @@ -169,6 +171,7 @@ func NewExecToolWithConfig( } return &ExecTool{ + cfg: cfg, workingDir: workingDir, timeout: timeout, denyPatterns: denyPatterns, @@ -270,12 +273,30 @@ func (t *ExecTool) executeRun(ctx context.Context, args map[string]any) *ToolRes // GHSA-pv8c-p6jf-3fpp: block exec from remote channels (e.g. Telegram webhooks) // unless explicitly opted-in via config. Fail-closed: empty channel = blocked. if !t.allowRemote { - channel := ToolChannel(ctx) - if channel == "" { - channel, _ = args["__channel"].(string) + channelName := ToolChannel(ctx) + if channelName == "" { + channelName, _ = args["__channel"].(string) } - channel = strings.TrimSpace(channel) - if channel == "" || !constants.IsInternalChannel(channel) { + channelName = strings.TrimSpace(channelName) + if channelName == "" { + return ErrorResult("exec is restricted to internal channels") + } + + var channelType string + if t.cfg != nil { + if channelConfig := t.cfg.Channels.Get(channelName); channelConfig != nil { + channelType = channelConfig.Type + } + } + // Fallback: if channelType is not determined from config, use channelName + if channelType == "" { + channelType = channelName + logger.DebugCF("shell", "Channel type not found in config, falling back to name", map[string]any{ + "channel_name": channelName, + }) + } + + if !constants.IsInternalChannel(channelType) { return ErrorResult("exec is restricted to internal channels") } }