Merge 6a7bf32f9e into 215d98aa78
This commit is contained in:
commit
95dfd083f2
33 changed files with 156 additions and 45 deletions
|
|
@ -536,7 +536,7 @@ func (al *AgentLoop) runAgentLoop(
|
||||||
// Record last channel for heartbeat notifications (skip internal channels and cli)
|
// Record last channel for heartbeat notifications (skip internal channels and cli)
|
||||||
if opts.Dispatch.Channel() != "" &&
|
if opts.Dispatch.Channel() != "" &&
|
||||||
opts.Dispatch.ChatID() != "" &&
|
opts.Dispatch.ChatID() != "" &&
|
||||||
!constants.IsInternalChannel(opts.Dispatch.Channel()) {
|
!constants.IsInternalChannel(opts.Dispatch.ChannelType()) {
|
||||||
channelKey := fmt.Sprintf("%s:%s", opts.Dispatch.Channel(), opts.Dispatch.ChatID())
|
channelKey := fmt.Sprintf("%s:%s", opts.Dispatch.Channel(), opts.Dispatch.ChatID())
|
||||||
if err := al.RecordLastChannel(channelKey); err != nil {
|
if err := al.RecordLastChannel(channelKey); err != nil {
|
||||||
logger.WarnCF(
|
logger.WarnCF(
|
||||||
|
|
|
||||||
|
|
@ -54,6 +54,7 @@ func (al *AgentLoop) ProcessDirectWithChannel(
|
||||||
msg := bus.InboundMessage{
|
msg := bus.InboundMessage{
|
||||||
Context: bus.InboundContext{
|
Context: bus.InboundContext{
|
||||||
Channel: channel,
|
Channel: channel,
|
||||||
|
ChannelType: channel, // For direct calls, channel name equals channel type
|
||||||
ChatID: chatID,
|
ChatID: chatID,
|
||||||
ChatType: "direct",
|
ChatType: "direct",
|
||||||
SenderID: "cron",
|
SenderID: "cron",
|
||||||
|
|
@ -87,6 +88,7 @@ func (al *AgentLoop) ProcessHeartbeat(
|
||||||
if channel != "" || chatID != "" {
|
if channel != "" || chatID != "" {
|
||||||
dispatch.InboundContext = &bus.InboundContext{
|
dispatch.InboundContext = &bus.InboundContext{
|
||||||
Channel: channel,
|
Channel: channel,
|
||||||
|
ChannelType: channel, // For heartbeat, channel name equals channel type
|
||||||
ChatID: chatID,
|
ChatID: chatID,
|
||||||
ChatType: "direct",
|
ChatType: "direct",
|
||||||
SenderID: "heartbeat",
|
SenderID: "heartbeat",
|
||||||
|
|
@ -256,7 +258,7 @@ func (al *AgentLoop) processSystemMessage(
|
||||||
// Parse origin channel from chat_id (format: "channel:chat_id")
|
// Parse origin channel from chat_id (format: "channel:chat_id")
|
||||||
var originChannel, originChatID string
|
var originChannel, originChatID string
|
||||||
if idx := strings.Index(msg.ChatID, ":"); idx > 0 {
|
if idx := strings.Index(msg.ChatID, ":"); idx > 0 {
|
||||||
originChannel = msg.ChatID[:idx]
|
originChannel = msg.ChatID[:idx] // e.g. "telegram"
|
||||||
originChatID = msg.ChatID[idx+1:]
|
originChatID = msg.ChatID[idx+1:]
|
||||||
} else {
|
} else {
|
||||||
originChannel = "cli"
|
originChannel = "cli"
|
||||||
|
|
|
||||||
|
|
@ -3987,7 +3987,8 @@ func TestProcessMessage_PicoPublishesReasoningAsThoughtMessage(t *testing.T) {
|
||||||
al := NewAgentLoop(cfg, msgBus, provider)
|
al := NewAgentLoop(cfg, msgBus, provider)
|
||||||
|
|
||||||
response, err := al.processMessage(context.Background(), bus.InboundMessage{
|
response, err := al.processMessage(context.Background(), bus.InboundMessage{
|
||||||
Channel: "pico",
|
Context: bus.InboundContext{Channel: "pico1", ChannelType: "pico"},
|
||||||
|
Channel: "pico1",
|
||||||
SenderID: "user1",
|
SenderID: "user1",
|
||||||
ChatID: "pico:test-session",
|
ChatID: "pico:test-session",
|
||||||
Content: "hello",
|
Content: "hello",
|
||||||
|
|
@ -4547,7 +4548,8 @@ func TestRun_PicoPublishesAssistantContentDuringToolCallsWithoutFinalDuplicate(t
|
||||||
}()
|
}()
|
||||||
|
|
||||||
if err := msgBus.PublishInbound(context.Background(), bus.InboundMessage{
|
if err := msgBus.PublishInbound(context.Background(), bus.InboundMessage{
|
||||||
Channel: "pico",
|
Context: bus.InboundContext{Channel: "pico1", ChannelType: "pico"},
|
||||||
|
Channel: "pico1",
|
||||||
SenderID: "user-1",
|
SenderID: "user-1",
|
||||||
ChatID: "session-1",
|
ChatID: "session-1",
|
||||||
Content: "run with tools",
|
Content: "run with tools",
|
||||||
|
|
|
||||||
|
|
@ -209,6 +209,9 @@ func appendEventContextFields(fields map[string]any, turnCtx *TurnContext) {
|
||||||
if inbound.Channel != "" {
|
if inbound.Channel != "" {
|
||||||
fields["inbound_channel"] = inbound.Channel
|
fields["inbound_channel"] = inbound.Channel
|
||||||
}
|
}
|
||||||
|
if inbound.ChannelType != "" {
|
||||||
|
fields["inbound_channel_type"] = inbound.ChannelType
|
||||||
|
}
|
||||||
if inbound.Account != "" {
|
if inbound.Account != "" {
|
||||||
fields["inbound_account"] = inbound.Account
|
fields["inbound_account"] = inbound.Account
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -27,6 +27,13 @@ func (r DispatchRequest) Channel() string {
|
||||||
return r.InboundContext.Channel
|
return r.InboundContext.Channel
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (r DispatchRequest) ChannelType() string {
|
||||||
|
if r.InboundContext == nil {
|
||||||
|
return ""
|
||||||
|
}
|
||||||
|
return r.InboundContext.ChannelType
|
||||||
|
}
|
||||||
|
|
||||||
func (r DispatchRequest) ChatID() string {
|
func (r DispatchRequest) ChatID() string {
|
||||||
if r.InboundContext == nil {
|
if r.InboundContext == nil {
|
||||||
return ""
|
return ""
|
||||||
|
|
@ -93,6 +100,10 @@ func normalizeProcessOptions(opts processOptions) processOptions {
|
||||||
MessageID: strings.TrimSpace(opts.MessageID),
|
MessageID: strings.TrimSpace(opts.MessageID),
|
||||||
ReplyToMessageID: strings.TrimSpace(opts.ReplyToMessageID),
|
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)
|
inbound.ChatType = inferChatTypeFromSessionScope(opts.Dispatch.SessionScope)
|
||||||
if inbound.Channel != "" || inbound.ChatID != "" || inbound.SenderID != "" ||
|
if inbound.Channel != "" || inbound.ChatID != "" || inbound.SenderID != "" ||
|
||||||
inbound.MessageID != "" || inbound.ReplyToMessageID != "" {
|
inbound.MessageID != "" || inbound.ReplyToMessageID != "" {
|
||||||
|
|
|
||||||
|
|
@ -10,6 +10,7 @@ import (
|
||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"github.com/sipeed/picoclaw/pkg/config"
|
||||||
"github.com/sipeed/picoclaw/pkg/constants"
|
"github.com/sipeed/picoclaw/pkg/constants"
|
||||||
runtimeevents "github.com/sipeed/picoclaw/pkg/events"
|
runtimeevents "github.com/sipeed/picoclaw/pkg/events"
|
||||||
"github.com/sipeed/picoclaw/pkg/logger"
|
"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(
|
al.bus.PublishOutbound(ctx, outboundMessageForTurn(
|
||||||
ts,
|
ts,
|
||||||
"Context window exceeded. Compressing history and retrying...",
|
"Context window exceeded. Compressing history and retrying...",
|
||||||
|
|
@ -433,11 +434,12 @@ func (p *Pipeline) CallLLM(
|
||||||
}
|
}
|
||||||
|
|
||||||
reasoningContent := responseReasoningContent(exec.response)
|
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 {
|
if shouldPublishPicoToolCallInterim {
|
||||||
// Pico tool-call turns publish their reasoning/content/tool summary as a
|
// Pico tool-call turns publish their reasoning/content/tool summary as a
|
||||||
// structured sequence after the tool-call payload is normalized below.
|
// 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)
|
go al.publishPicoReasoning(turnCtx, reasoningContent, ts.chatID)
|
||||||
} else {
|
} else {
|
||||||
go al.handleReasoning(
|
go al.handleReasoning(
|
||||||
|
|
@ -476,7 +478,8 @@ func (p *Pipeline) CallLLM(
|
||||||
// No-tool-call path: steering check and direct response
|
// No-tool-call path: steering check and direct response
|
||||||
if len(exec.response.ToolCalls) == 0 || exec.gracefulTerminal {
|
if len(exec.response.ToolCalls) == 0 || exec.gracefulTerminal {
|
||||||
responseContent := exec.response.Content
|
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
|
responseContent = exec.response.ReasoningContent
|
||||||
}
|
}
|
||||||
if steerMsgs := al.dequeueSteeringMessagesForScope(ts.sessionKey); len(steerMsgs) > 0 {
|
if steerMsgs := al.dequeueSteeringMessagesForScope(ts.sessionKey); len(steerMsgs) > 0 {
|
||||||
|
|
|
||||||
|
|
@ -239,6 +239,7 @@ func (a *Agent) processUtterance(ctx context.Context, acc *speechAccumulator) {
|
||||||
if err := a.bus.PublishInbound(ctx, bus.InboundMessage{
|
if err := a.bus.PublishInbound(ctx, bus.InboundMessage{
|
||||||
Context: bus.InboundContext{
|
Context: bus.InboundContext{
|
||||||
Channel: channelType,
|
Channel: channelType,
|
||||||
|
ChannelType: channelType,
|
||||||
ChatID: acc.chatID,
|
ChatID: acc.chatID,
|
||||||
ChatType: "channel",
|
ChatType: "channel",
|
||||||
SenderID: acc.speakerID,
|
SenderID: acc.speakerID,
|
||||||
|
|
|
||||||
|
|
@ -32,6 +32,7 @@ func NormalizeInboundMessage(msg InboundMessage) InboundMessage {
|
||||||
|
|
||||||
func (ctx InboundContext) isZero() bool {
|
func (ctx InboundContext) isZero() bool {
|
||||||
return ctx.Channel == "" &&
|
return ctx.Channel == "" &&
|
||||||
|
ctx.ChannelType == "" &&
|
||||||
ctx.Account == "" &&
|
ctx.Account == "" &&
|
||||||
ctx.ChatID == "" &&
|
ctx.ChatID == "" &&
|
||||||
ctx.ChatType == "" &&
|
ctx.ChatType == "" &&
|
||||||
|
|
@ -49,6 +50,11 @@ func (ctx InboundContext) isZero() bool {
|
||||||
|
|
||||||
func normalizeInboundContext(ctx InboundContext) InboundContext {
|
func normalizeInboundContext(ctx InboundContext) InboundContext {
|
||||||
ctx.Channel = strings.TrimSpace(ctx.Channel)
|
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.Account = strings.TrimSpace(ctx.Account)
|
||||||
ctx.ChatID = strings.TrimSpace(ctx.ChatID)
|
ctx.ChatID = strings.TrimSpace(ctx.ChatID)
|
||||||
ctx.ChatType = normalizeKind(ctx.ChatType)
|
ctx.ChatType = normalizeKind(ctx.ChatType)
|
||||||
|
|
|
||||||
|
|
@ -14,6 +14,7 @@ type SenderInfo struct {
|
||||||
// allocation.
|
// allocation.
|
||||||
type InboundContext struct {
|
type InboundContext struct {
|
||||||
Channel string `json:"channel"`
|
Channel string `json:"channel"`
|
||||||
|
ChannelType string `json:"channel_type,omitempty"` // telegram, discord, slack, etc.
|
||||||
Account string `json:"account,omitempty"`
|
Account string `json:"account,omitempty"`
|
||||||
|
|
||||||
ChatID string `json:"chat_id"`
|
ChatID string `json:"chat_id"`
|
||||||
|
|
|
||||||
|
|
@ -75,6 +75,11 @@ func WithReasoningChannelID(id string) BaseChannelOption {
|
||||||
return func(c *BaseChannel) { c.reasoningChannelID = id }
|
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
|
// MessageLengthProvider is an opt-in interface that channels implement
|
||||||
// to advertise their maximum message length. The Manager uses this via
|
// to advertise their maximum message length. The Manager uses this via
|
||||||
// type assertion to decide whether to split outbound messages.
|
// type assertion to decide whether to split outbound messages.
|
||||||
|
|
@ -87,6 +92,7 @@ type BaseChannel struct {
|
||||||
bus *bus.MessageBus
|
bus *bus.MessageBus
|
||||||
running atomic.Bool
|
running atomic.Bool
|
||||||
name string
|
name string
|
||||||
|
channelType string
|
||||||
allowList []string
|
allowList []string
|
||||||
maxMessageLength int
|
maxMessageLength int
|
||||||
groupTrigger config.GroupTriggerConfig
|
groupTrigger config.GroupTriggerConfig
|
||||||
|
|
@ -187,6 +193,11 @@ func (c *BaseChannel) Name() string {
|
||||||
return c.name
|
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
|
// 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).
|
// to ensure the name matches the config key (which may differ from the type).
|
||||||
func (c *BaseChannel) SetName(name string) {
|
func (c *BaseChannel) SetName(name string) {
|
||||||
|
|
@ -294,6 +305,7 @@ func (c *BaseChannel) HandleMessageWithContext(
|
||||||
}
|
}
|
||||||
|
|
||||||
inboundCtx.Channel = c.name
|
inboundCtx.Channel = c.name
|
||||||
|
inboundCtx.ChannelType = c.channelType
|
||||||
if inboundCtx.ChatID == "" {
|
if inboundCtx.ChatID == "" {
|
||||||
inboundCtx.ChatID = deliveryChatID
|
inboundCtx.ChatID = deliveryChatID
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -52,6 +52,7 @@ func NewDingTalkChannel(
|
||||||
channels.WithMaxMessageLength(20000),
|
channels.WithMaxMessageLength(20000),
|
||||||
channels.WithGroupTrigger(bc.GroupTrigger),
|
channels.WithGroupTrigger(bc.GroupTrigger),
|
||||||
channels.WithReasoningChannelID(bc.ReasoningChannelID),
|
channels.WithReasoningChannelID(bc.ReasoningChannelID),
|
||||||
|
channels.WithChannelType(bc.Type),
|
||||||
)
|
)
|
||||||
|
|
||||||
return &DingTalkChannel{
|
return &DingTalkChannel{
|
||||||
|
|
|
||||||
|
|
@ -85,6 +85,7 @@ func NewDiscordChannel(
|
||||||
channels.WithMaxMessageLength(2000),
|
channels.WithMaxMessageLength(2000),
|
||||||
channels.WithGroupTrigger(bc.GroupTrigger),
|
channels.WithGroupTrigger(bc.GroupTrigger),
|
||||||
channels.WithReasoningChannelID(bc.ReasoningChannelID),
|
channels.WithReasoningChannelID(bc.ReasoningChannelID),
|
||||||
|
channels.WithChannelType(bc.Type),
|
||||||
)
|
)
|
||||||
|
|
||||||
ch := &DiscordChannel{
|
ch := &DiscordChannel{
|
||||||
|
|
@ -669,7 +670,6 @@ func (c *DiscordChannel) handleMessage(s *discordgo.Session, m *discordgo.Messag
|
||||||
"is_dm": fmt.Sprintf("%t", m.GuildID == ""),
|
"is_dm": fmt.Sprintf("%t", m.GuildID == ""),
|
||||||
}
|
}
|
||||||
inboundCtx := bus.InboundContext{
|
inboundCtx := bus.InboundContext{
|
||||||
Channel: c.Name(),
|
|
||||||
ChatID: m.ChannelID,
|
ChatID: m.ChannelID,
|
||||||
ChatType: peerKind,
|
ChatType: peerKind,
|
||||||
SenderID: senderID,
|
SenderID: senderID,
|
||||||
|
|
|
||||||
|
|
@ -63,6 +63,7 @@ func NewFeishuChannel(bc *config.Channel, cfg *config.FeishuSettings, bus *bus.M
|
||||||
base := channels.NewBaseChannel("feishu", cfg, bus, bc.AllowFrom,
|
base := channels.NewBaseChannel("feishu", cfg, bus, bc.AllowFrom,
|
||||||
channels.WithGroupTrigger(bc.GroupTrigger),
|
channels.WithGroupTrigger(bc.GroupTrigger),
|
||||||
channels.WithReasoningChannelID(bc.ReasoningChannelID),
|
channels.WithReasoningChannelID(bc.ReasoningChannelID),
|
||||||
|
channels.WithChannelType(bc.Type),
|
||||||
)
|
)
|
||||||
|
|
||||||
tc := newTokenCache()
|
tc := newTokenCache()
|
||||||
|
|
|
||||||
|
|
@ -38,6 +38,7 @@ func NewIRCChannel(bc *config.Channel, cfg *config.IRCSettings, messageBus *bus.
|
||||||
channels.WithMaxMessageLength(400),
|
channels.WithMaxMessageLength(400),
|
||||||
channels.WithGroupTrigger(bc.GroupTrigger),
|
channels.WithGroupTrigger(bc.GroupTrigger),
|
||||||
channels.WithReasoningChannelID(bc.ReasoningChannelID),
|
channels.WithReasoningChannelID(bc.ReasoningChannelID),
|
||||||
|
channels.WithChannelType(bc.Type),
|
||||||
)
|
)
|
||||||
|
|
||||||
return &IRCChannel{
|
return &IRCChannel{
|
||||||
|
|
|
||||||
|
|
@ -73,6 +73,7 @@ func NewLINEChannel(
|
||||||
channels.WithMaxMessageLength(5000),
|
channels.WithMaxMessageLength(5000),
|
||||||
channels.WithGroupTrigger(bc.GroupTrigger),
|
channels.WithGroupTrigger(bc.GroupTrigger),
|
||||||
channels.WithReasoningChannelID(bc.ReasoningChannelID),
|
channels.WithReasoningChannelID(bc.ReasoningChannelID),
|
||||||
|
channels.WithChannelType(bc.Type),
|
||||||
)
|
)
|
||||||
|
|
||||||
return &LINEChannel{
|
return &LINEChannel{
|
||||||
|
|
|
||||||
|
|
@ -43,6 +43,7 @@ func NewMaixCamChannel(
|
||||||
bus,
|
bus,
|
||||||
bc.AllowFrom,
|
bc.AllowFrom,
|
||||||
channels.WithReasoningChannelID(bc.ReasoningChannelID),
|
channels.WithReasoningChannelID(bc.ReasoningChannelID),
|
||||||
|
channels.WithChannelType(bc.Type),
|
||||||
)
|
)
|
||||||
|
|
||||||
return &MaixCamChannel{
|
return &MaixCamChannel{
|
||||||
|
|
|
||||||
|
|
@ -155,6 +155,10 @@ func outboundMessageChannel(msg bus.OutboundMessage) string {
|
||||||
return msg.Context.Channel
|
return msg.Context.Channel
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func outboundMessageChannelType(msg bus.OutboundMessage) string {
|
||||||
|
return msg.Context.ChannelType
|
||||||
|
}
|
||||||
|
|
||||||
func outboundMessageChatID(msg bus.OutboundMessage) string {
|
func outboundMessageChatID(msg bus.OutboundMessage) string {
|
||||||
return msg.ChatID
|
return msg.ChatID
|
||||||
}
|
}
|
||||||
|
|
@ -178,6 +182,10 @@ func outboundMediaChannel(msg bus.OutboundMediaMessage) string {
|
||||||
return msg.Context.Channel
|
return msg.Context.Channel
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func outboundMediaChannelType(msg bus.OutboundMediaMessage) string {
|
||||||
|
return msg.Context.ChannelType
|
||||||
|
}
|
||||||
|
|
||||||
func outboundMediaChatID(msg bus.OutboundMediaMessage) string {
|
func outboundMediaChatID(msg bus.OutboundMediaMessage) string {
|
||||||
return msg.ChatID
|
return msg.ChatID
|
||||||
}
|
}
|
||||||
|
|
@ -1200,6 +1208,7 @@ func dispatchLoop[M any](
|
||||||
m *Manager,
|
m *Manager,
|
||||||
ch <-chan M,
|
ch <-chan M,
|
||||||
getChannel func(M) string,
|
getChannel func(M) string,
|
||||||
|
getChannelType func(M) string,
|
||||||
enqueue func(context.Context, *channelWorker, M) bool,
|
enqueue func(context.Context, *channelWorker, M) bool,
|
||||||
startMsg, stopMsg, unknownMsg, noWorkerMsg string,
|
startMsg, stopMsg, unknownMsg, noWorkerMsg string,
|
||||||
) {
|
) {
|
||||||
|
|
@ -1218,9 +1227,10 @@ func dispatchLoop[M any](
|
||||||
}
|
}
|
||||||
|
|
||||||
channel := getChannel(msg)
|
channel := getChannel(msg)
|
||||||
|
channelType := getChannelType(msg)
|
||||||
|
|
||||||
// Silently skip internal channels
|
// Silently skip internal channels
|
||||||
if constants.IsInternalChannel(channel) {
|
if constants.IsInternalChannel(channelType) {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -1250,6 +1260,7 @@ func (m *Manager) dispatchOutbound(ctx context.Context) {
|
||||||
ctx, m,
|
ctx, m,
|
||||||
m.bus.OutboundChan(),
|
m.bus.OutboundChan(),
|
||||||
func(msg bus.OutboundMessage) string { return outboundMessageChannel(msg) },
|
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 {
|
func(ctx context.Context, w *channelWorker, msg bus.OutboundMessage) bool {
|
||||||
select {
|
select {
|
||||||
case w.queue <- msg:
|
case w.queue <- msg:
|
||||||
|
|
@ -1271,6 +1282,7 @@ func (m *Manager) dispatchOutboundMedia(ctx context.Context) {
|
||||||
ctx, m,
|
ctx, m,
|
||||||
m.bus.OutboundMediaChan(),
|
m.bus.OutboundMediaChan(),
|
||||||
func(msg bus.OutboundMediaMessage) string { return outboundMediaChannel(msg) },
|
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 {
|
func(ctx context.Context, w *channelWorker, msg bus.OutboundMediaMessage) bool {
|
||||||
select {
|
select {
|
||||||
case w.mediaQueue <- msg:
|
case w.mediaQueue <- msg:
|
||||||
|
|
|
||||||
|
|
@ -242,6 +242,7 @@ func NewMatrixChannel(
|
||||||
channels.WithMaxMessageLength(65536),
|
channels.WithMaxMessageLength(65536),
|
||||||
channels.WithGroupTrigger(bc.GroupTrigger),
|
channels.WithGroupTrigger(bc.GroupTrigger),
|
||||||
channels.WithReasoningChannelID(bc.ReasoningChannelID),
|
channels.WithReasoningChannelID(bc.ReasoningChannelID),
|
||||||
|
channels.WithChannelType(bc.Type),
|
||||||
)
|
)
|
||||||
|
|
||||||
ch := &MatrixChannel{
|
ch := &MatrixChannel{
|
||||||
|
|
|
||||||
|
|
@ -104,6 +104,7 @@ func NewOneBotChannel(
|
||||||
base := channels.NewBaseChannel("onebot", cfg, messageBus, bc.AllowFrom,
|
base := channels.NewBaseChannel("onebot", cfg, messageBus, bc.AllowFrom,
|
||||||
channels.WithGroupTrigger(bc.GroupTrigger),
|
channels.WithGroupTrigger(bc.GroupTrigger),
|
||||||
channels.WithReasoningChannelID(bc.ReasoningChannelID),
|
channels.WithReasoningChannelID(bc.ReasoningChannelID),
|
||||||
|
channels.WithChannelType(bc.Type),
|
||||||
)
|
)
|
||||||
|
|
||||||
const dedupSize = 1024
|
const dedupSize = 1024
|
||||||
|
|
|
||||||
|
|
@ -39,7 +39,9 @@ func NewPicoClientChannel(
|
||||||
return nil, fmt.Errorf("pico_client url is required")
|
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{
|
return &PicoClientChannel{
|
||||||
BaseChannel: base,
|
BaseChannel: base,
|
||||||
|
|
|
||||||
|
|
@ -117,7 +117,9 @@ func NewPicoChannel(
|
||||||
return nil, fmt.Errorf("pico token is required")
|
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
|
allowOrigins := cfg.AllowOrigins
|
||||||
checkOrigin := func(r *http.Request) bool {
|
checkOrigin := func(r *http.Request) bool {
|
||||||
|
|
@ -965,7 +967,8 @@ func (c *PicoChannel) handleMessageSend(pc *picoConn, msg PicoMessage) {
|
||||||
}
|
}
|
||||||
|
|
||||||
inboundCtx := bus.InboundContext{
|
inboundCtx := bus.InboundContext{
|
||||||
Channel: "pico",
|
Channel: c.bc.Name(),
|
||||||
|
ChannelType: config.ChannelPico,
|
||||||
ChatID: chatID,
|
ChatID: chatID,
|
||||||
ChatType: "direct",
|
ChatType: "direct",
|
||||||
SenderID: senderID,
|
SenderID: senderID,
|
||||||
|
|
|
||||||
|
|
@ -88,6 +88,7 @@ func NewQQChannel(bc *config.Channel, cfg *config.QQSettings, messageBus *bus.Me
|
||||||
channels.WithMaxMessageLength(cfg.MaxMessageLength),
|
channels.WithMaxMessageLength(cfg.MaxMessageLength),
|
||||||
channels.WithGroupTrigger(bc.GroupTrigger),
|
channels.WithGroupTrigger(bc.GroupTrigger),
|
||||||
channels.WithReasoningChannelID(bc.ReasoningChannelID),
|
channels.WithReasoningChannelID(bc.ReasoningChannelID),
|
||||||
|
channels.WithChannelType(bc.Type),
|
||||||
)
|
)
|
||||||
|
|
||||||
return &QQChannel{
|
return &QQChannel{
|
||||||
|
|
|
||||||
|
|
@ -56,6 +56,7 @@ func NewSlackChannel(
|
||||||
channels.WithMaxMessageLength(40000),
|
channels.WithMaxMessageLength(40000),
|
||||||
channels.WithGroupTrigger(bc.GroupTrigger),
|
channels.WithGroupTrigger(bc.GroupTrigger),
|
||||||
channels.WithReasoningChannelID(bc.ReasoningChannelID),
|
channels.WithReasoningChannelID(bc.ReasoningChannelID),
|
||||||
|
channels.WithChannelType(bc.Type),
|
||||||
)
|
)
|
||||||
|
|
||||||
return &SlackChannel{
|
return &SlackChannel{
|
||||||
|
|
@ -381,7 +382,6 @@ func (c *SlackChannel) handleMessageEvent(ev *slackevents.MessageEvent) {
|
||||||
})
|
})
|
||||||
|
|
||||||
inboundCtx := bus.InboundContext{
|
inboundCtx := bus.InboundContext{
|
||||||
Channel: c.Name(),
|
|
||||||
Account: c.teamID,
|
Account: c.teamID,
|
||||||
ChatID: channelID,
|
ChatID: channelID,
|
||||||
ChatType: peerKind,
|
ChatType: peerKind,
|
||||||
|
|
@ -456,7 +456,6 @@ func (c *SlackChannel) handleAppMention(ev *slackevents.AppMentionEvent) {
|
||||||
"team_id": c.teamID,
|
"team_id": c.teamID,
|
||||||
}
|
}
|
||||||
inboundCtx := bus.InboundContext{
|
inboundCtx := bus.InboundContext{
|
||||||
Channel: c.Name(),
|
|
||||||
Account: c.teamID,
|
Account: c.teamID,
|
||||||
ChatID: channelID,
|
ChatID: channelID,
|
||||||
ChatType: mentionPeerKind,
|
ChatType: mentionPeerKind,
|
||||||
|
|
@ -521,7 +520,6 @@ func (c *SlackChannel) handleSlashCommand(event socketmode.Event) {
|
||||||
peerKind = "direct"
|
peerKind = "direct"
|
||||||
}
|
}
|
||||||
inboundCtx := bus.InboundContext{
|
inboundCtx := bus.InboundContext{
|
||||||
Channel: c.Name(),
|
|
||||||
Account: c.teamID,
|
Account: c.teamID,
|
||||||
ChatID: channelID,
|
ChatID: channelID,
|
||||||
ChatType: peerKind,
|
ChatType: peerKind,
|
||||||
|
|
|
||||||
|
|
@ -95,6 +95,7 @@ func NewTeamsWebhookChannel(
|
||||||
"*",
|
"*",
|
||||||
}, // Output-only channel; "*" suppresses misleading "allows EVERYONE" audit warning
|
}, // Output-only channel; "*" suppresses misleading "allows EVERYONE" audit warning
|
||||||
channels.WithMaxMessageLength(24000), // Power Automate webhook payload limit is 28KB
|
channels.WithMaxMessageLength(24000), // Power Automate webhook payload limit is 28KB
|
||||||
|
channels.WithChannelType(bc.Type),
|
||||||
)
|
)
|
||||||
|
|
||||||
client := goteamsnotify.NewTeamsClient()
|
client := goteamsnotify.NewTeamsClient()
|
||||||
|
|
|
||||||
|
|
@ -123,6 +123,7 @@ func NewTelegramChannel(
|
||||||
channels.WithMaxMessageLength(4000),
|
channels.WithMaxMessageLength(4000),
|
||||||
channels.WithGroupTrigger(bc.GroupTrigger),
|
channels.WithGroupTrigger(bc.GroupTrigger),
|
||||||
channels.WithReasoningChannelID(bc.ReasoningChannelID),
|
channels.WithReasoningChannelID(bc.ReasoningChannelID),
|
||||||
|
channels.WithChannelType(bc.Type),
|
||||||
)
|
)
|
||||||
|
|
||||||
ch := &TelegramChannel{
|
ch := &TelegramChannel{
|
||||||
|
|
@ -1005,7 +1006,6 @@ func (c *TelegramChannel) handleMessages(ctx context.Context, messages []*telego
|
||||||
}
|
}
|
||||||
|
|
||||||
inboundCtx := bus.InboundContext{
|
inboundCtx := bus.InboundContext{
|
||||||
Channel: c.Name(),
|
|
||||||
ChatID: fmt.Sprintf("%d", chatID),
|
ChatID: fmt.Sprintf("%d", chatID),
|
||||||
ChatType: peerKind,
|
ChatType: peerKind,
|
||||||
SenderID: platformID,
|
SenderID: platformID,
|
||||||
|
|
|
||||||
|
|
@ -45,6 +45,7 @@ func NewVKChannel(channelName string, bc *config.Channel, bus *bus.MessageBus) (
|
||||||
channels.WithMaxMessageLength(4000),
|
channels.WithMaxMessageLength(4000),
|
||||||
channels.WithGroupTrigger(bc.GroupTrigger),
|
channels.WithGroupTrigger(bc.GroupTrigger),
|
||||||
channels.WithReasoningChannelID(bc.ReasoningChannelID),
|
channels.WithReasoningChannelID(bc.ReasoningChannelID),
|
||||||
|
channels.WithChannelType(bc.Type),
|
||||||
)
|
)
|
||||||
|
|
||||||
return &VKChannel{
|
return &VKChannel{
|
||||||
|
|
|
||||||
|
|
@ -122,6 +122,7 @@ func NewChannel(bc *config.Channel, cfg *config.WeComSettings, messageBus *bus.M
|
||||||
messageBus,
|
messageBus,
|
||||||
bc.AllowFrom,
|
bc.AllowFrom,
|
||||||
channels.WithReasoningChannelID(bc.ReasoningChannelID),
|
channels.WithReasoningChannelID(bc.ReasoningChannelID),
|
||||||
|
channels.WithChannelType(bc.Type),
|
||||||
)
|
)
|
||||||
|
|
||||||
ch := &WeComChannel{
|
ch := &WeComChannel{
|
||||||
|
|
|
||||||
|
|
@ -78,6 +78,7 @@ func NewWeixinChannel(
|
||||||
bc.AllowFrom,
|
bc.AllowFrom,
|
||||||
channels.WithMaxMessageLength(4000),
|
channels.WithMaxMessageLength(4000),
|
||||||
channels.WithReasoningChannelID(bc.ReasoningChannelID),
|
channels.WithReasoningChannelID(bc.ReasoningChannelID),
|
||||||
|
channels.WithChannelType(bc.Type),
|
||||||
)
|
)
|
||||||
|
|
||||||
return &WeixinChannel{
|
return &WeixinChannel{
|
||||||
|
|
|
||||||
|
|
@ -40,6 +40,7 @@ func NewWhatsAppChannel(
|
||||||
bc.AllowFrom,
|
bc.AllowFrom,
|
||||||
channels.WithMaxMessageLength(65536),
|
channels.WithMaxMessageLength(65536),
|
||||||
channels.WithReasoningChannelID(bc.ReasoningChannelID),
|
channels.WithReasoningChannelID(bc.ReasoningChannelID),
|
||||||
|
channels.WithChannelType(bc.Type),
|
||||||
)
|
)
|
||||||
|
|
||||||
return &WhatsAppChannel{
|
return &WhatsAppChannel{
|
||||||
|
|
|
||||||
|
|
@ -70,7 +70,10 @@ func NewWhatsAppNativeChannel(
|
||||||
bus *bus.MessageBus,
|
bus *bus.MessageBus,
|
||||||
storePath string,
|
storePath string,
|
||||||
) (channels.Channel, error) {
|
) (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 == "" {
|
if storePath == "" {
|
||||||
storePath = "whatsapp"
|
storePath = "whatsapp"
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -10,7 +10,7 @@ var internalChannels = map[string]struct{}{
|
||||||
}
|
}
|
||||||
|
|
||||||
// IsInternalChannel returns true if the channel is an internal channel.
|
// IsInternalChannel returns true if the channel is an internal channel.
|
||||||
func IsInternalChannel(channel string) bool {
|
func IsInternalChannel(channelType string) bool {
|
||||||
_, found := internalChannels[channel]
|
_, found := internalChannels[channelType]
|
||||||
return found
|
return found
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -12,6 +12,7 @@ import (
|
||||||
"github.com/sipeed/picoclaw/pkg/config"
|
"github.com/sipeed/picoclaw/pkg/config"
|
||||||
"github.com/sipeed/picoclaw/pkg/constants"
|
"github.com/sipeed/picoclaw/pkg/constants"
|
||||||
"github.com/sipeed/picoclaw/pkg/cron"
|
"github.com/sipeed/picoclaw/pkg/cron"
|
||||||
|
"github.com/sipeed/picoclaw/pkg/logger"
|
||||||
"github.com/sipeed/picoclaw/pkg/utils"
|
"github.com/sipeed/picoclaw/pkg/utils"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
@ -25,6 +26,7 @@ type JobExecutor interface {
|
||||||
|
|
||||||
// CronTool provides scheduling capabilities for the agent
|
// CronTool provides scheduling capabilities for the agent
|
||||||
type CronTool struct {
|
type CronTool struct {
|
||||||
|
cfg *config.Config
|
||||||
cronService *cron.CronService
|
cronService *cron.CronService
|
||||||
executor JobExecutor
|
executor JobExecutor
|
||||||
msgBus *bus.MessageBus
|
msgBus *bus.MessageBus
|
||||||
|
|
@ -59,6 +61,7 @@ func NewCronTool(
|
||||||
execTool.SetTimeout(execTimeout)
|
execTool.SetTimeout(execTimeout)
|
||||||
}
|
}
|
||||||
return &CronTool{
|
return &CronTool{
|
||||||
|
cfg: config,
|
||||||
cronService: cronService,
|
cronService: cronService,
|
||||||
executor: executor,
|
executor: executor,
|
||||||
msgBus: msgBus,
|
msgBus: msgBus,
|
||||||
|
|
@ -201,7 +204,22 @@ func (t *CronTool) addJob(ctx context.Context, args map[string]any) *ToolResult
|
||||||
if !t.execEnabled {
|
if !t.execEnabled {
|
||||||
return ErrorResult("command execution is disabled")
|
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")
|
return ErrorResult("scheduling command execution is restricted to internal channels")
|
||||||
}
|
}
|
||||||
if !t.allowCommand && !commandConfirm {
|
if !t.allowCommand && !commandConfirm {
|
||||||
|
|
|
||||||
|
|
@ -21,6 +21,7 @@ import (
|
||||||
"github.com/sipeed/picoclaw/pkg/config"
|
"github.com/sipeed/picoclaw/pkg/config"
|
||||||
"github.com/sipeed/picoclaw/pkg/constants"
|
"github.com/sipeed/picoclaw/pkg/constants"
|
||||||
"github.com/sipeed/picoclaw/pkg/isolation"
|
"github.com/sipeed/picoclaw/pkg/isolation"
|
||||||
|
"github.com/sipeed/picoclaw/pkg/logger"
|
||||||
)
|
)
|
||||||
|
|
||||||
var (
|
var (
|
||||||
|
|
@ -35,6 +36,7 @@ func getSessionManager() *SessionManager {
|
||||||
}
|
}
|
||||||
|
|
||||||
type ExecTool struct {
|
type ExecTool struct {
|
||||||
|
cfg *config.Config
|
||||||
workingDir string
|
workingDir string
|
||||||
timeout time.Duration
|
timeout time.Duration
|
||||||
denyPatterns []*regexp.Regexp
|
denyPatterns []*regexp.Regexp
|
||||||
|
|
@ -169,6 +171,7 @@ func NewExecToolWithConfig(
|
||||||
}
|
}
|
||||||
|
|
||||||
return &ExecTool{
|
return &ExecTool{
|
||||||
|
cfg: cfg,
|
||||||
workingDir: workingDir,
|
workingDir: workingDir,
|
||||||
timeout: timeout,
|
timeout: timeout,
|
||||||
denyPatterns: denyPatterns,
|
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)
|
// GHSA-pv8c-p6jf-3fpp: block exec from remote channels (e.g. Telegram webhooks)
|
||||||
// unless explicitly opted-in via config. Fail-closed: empty channel = blocked.
|
// unless explicitly opted-in via config. Fail-closed: empty channel = blocked.
|
||||||
if !t.allowRemote {
|
if !t.allowRemote {
|
||||||
channel := ToolChannel(ctx)
|
channelName := ToolChannel(ctx)
|
||||||
if channel == "" {
|
if channelName == "" {
|
||||||
channel, _ = args["__channel"].(string)
|
channelName, _ = args["__channel"].(string)
|
||||||
}
|
}
|
||||||
channel = strings.TrimSpace(channel)
|
channelName = strings.TrimSpace(channelName)
|
||||||
if channel == "" || !constants.IsInternalChannel(channel) {
|
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")
|
return ErrorResult("exec is restricted to internal channels")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue