diff --git a/pkg/agent/agent.go b/pkg/agent/agent.go index 5749149c1..a94d58c4c 100644 --- a/pkg/agent/agent.go +++ b/pkg/agent/agent.go @@ -536,7 +536,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 4d2886a80..f8569f5c6 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{ @@ -256,7 +258,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 7a869ec94..d864beccd 100644 --- a/pkg/agent/agent_test.go +++ b/pkg/agent/agent_test.go @@ -3987,7 +3987,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", @@ -4547,7 +4548,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 9228b6d55..0ae9139d2 100644 --- a/pkg/agent/agent_utils.go +++ b/pkg/agent/agent_utils.go @@ -209,6 +209,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 3a3c496f6..e0f4ae6f2 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" runtimeevents "github.com/sipeed/picoclaw/pkg/events" "github.com/sipeed/picoclaw/pkg/logger" @@ -338,7 +339,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...", @@ -433,11 +434,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( @@ -476,7 +478,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 d09c021c7..2e9fbfe65 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 d4d34211d..9dce1823e 100644 --- a/pkg/channels/line/line.go +++ b/pkg/channels/line/line.go @@ -73,6 +73,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 d345a5d0b..2294e88bc 100644 --- a/pkg/channels/manager.go +++ b/pkg/channels/manager.go @@ -155,6 +155,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 } @@ -178,6 +182,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 } @@ -1200,6 +1208,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, ) { @@ -1218,9 +1227,10 @@ func dispatchLoop[M any]( } channel := getChannel(msg) + channelType := getChannelType(msg) // Silently skip internal channels - if constants.IsInternalChannel(channel) { + if constants.IsInternalChannel(channelType) { continue } @@ -1250,6 +1260,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: @@ -1271,6 +1282,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 0965bcedc..842709124 100644 --- a/pkg/channels/telegram/telegram.go +++ b/pkg/channels/telegram/telegram.go @@ -123,6 +123,7 @@ func NewTelegramChannel( channels.WithMaxMessageLength(4000), channels.WithGroupTrigger(bc.GroupTrigger), channels.WithReasoningChannelID(bc.ReasoningChannelID), + channels.WithChannelType(bc.Type), ) ch := &TelegramChannel{ @@ -1005,7 +1006,6 @@ func (c *TelegramChannel) handleMessages(ctx context.Context, messages []*telego } 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 a9547eba9..0439d8fbe 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") } }