diff --git a/cmd/gateway.go b/cmd/gateway.go index eacd86fc..b65d559a 100644 --- a/cmd/gateway.go +++ b/cmd/gateway.go @@ -891,7 +891,7 @@ func runGateway() { defer sched.Stop() // Start cron service with job handler (routes through scheduler's cron lane) - pgStores.Cron.SetOnJob(makeCronJobHandler(sched, msgBus, cfg)) + pgStores.Cron.SetOnJob(makeCronJobHandler(sched, msgBus, cfg, channelMgr)) if err := pgStores.Cron.Start(); err != nil { slog.Warn("cron service failed to start", "error", err) } diff --git a/cmd/gateway_consumer.go b/cmd/gateway_consumer.go index da3c4543..c8508b24 100644 --- a/cmd/gateway_consumer.go +++ b/cmd/gateway_consumer.go @@ -249,6 +249,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents Media: reqMedia, ForwardMedia: fwdMedia, Channel: msg.Channel, + ChannelType: resolveChannelType(channelMgr, msg.Channel), ChatID: msg.ChatID, PeerKind: peerKind, LocalKey: msg.Metadata["local_key"], @@ -389,6 +390,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents origChannel := msg.Metadata["origin_channel"] origPeerKind := msg.Metadata["origin_peer_kind"] origLocalKey := msg.Metadata["origin_local_key"] + origChannelType := resolveChannelType(channelMgr, origChannel) parentAgent := msg.Metadata["parent_agent"] if parentAgent == "" { parentAgent = "default" @@ -451,6 +453,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents ForwardMedia: fwdMedia, ContentSuffix: contentSuffix, Channel: origChannel, + ChannelType: origChannelType, ChatID: msg.ChatID, PeerKind: origPeerKind, LocalKey: origLocalKey, @@ -526,6 +529,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents origChannel := msg.Metadata["origin_channel"] origPeerKind := msg.Metadata["origin_peer_kind"] origLocalKey := msg.Metadata["origin_local_key"] + origChannelType := resolveChannelType(channelMgr, origChannel) parentAgent := msg.Metadata["parent_agent"] if parentAgent == "" { parentAgent = "default" @@ -585,6 +589,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents ForwardMedia: fwdMedia, ContentSuffix: contentSuffix, Channel: origChannel, + ChannelType: origChannelType, ChatID: msg.ChatID, PeerKind: origPeerKind, LocalKey: origLocalKey, @@ -653,6 +658,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents origChannel := msg.Metadata["origin_channel"] origPeerKind := msg.Metadata["origin_peer_kind"] origLocalKey := msg.Metadata["origin_local_key"] + origChannelType := resolveChannelType(channelMgr, origChannel) targetAgent := msg.AgentID if targetAgent == "" { targetAgent = cfg.ResolveDefaultAgentID() @@ -683,15 +689,16 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents outMeta := buildAnnounceOutMeta(origLocalKey) outCh := sched.Schedule(ctx, scheduler.LaneDelegate, agent.RunRequest{ - SessionKey: sessionKey, - Message: msg.Content, - Channel: origChannel, - ChatID: msg.ChatID, - PeerKind: origPeerKind, - LocalKey: origLocalKey, - UserID: announceUserID, - RunID: fmt.Sprintf("handoff-%s", msg.Metadata["handoff_id"]), - Stream: false, + SessionKey: sessionKey, + Message: msg.Content, + Channel: origChannel, + ChannelType: origChannelType, + ChatID: msg.ChatID, + PeerKind: origPeerKind, + LocalKey: origLocalKey, + UserID: announceUserID, + RunID: fmt.Sprintf("handoff-%s", msg.Metadata["handoff_id"]), + Stream: false, }) go func(origCh, chatID string, meta map[string]string) { @@ -719,6 +726,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents origChannel := msg.Metadata["origin_channel"] origPeerKind := msg.Metadata["origin_peer_kind"] origLocalKey := msg.Metadata["origin_local_key"] + origChannelType := resolveChannelType(channelMgr, origChannel) targetAgent := msg.AgentID // team_message sets AgentID to the target agent key if targetAgent == "" { targetAgent = cfg.ResolveDefaultAgentID() @@ -749,15 +757,16 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents outMeta := buildAnnounceOutMeta(origLocalKey) outCh := sched.Schedule(ctx, scheduler.LaneDelegate, agent.RunRequest{ - SessionKey: sessionKey, - Message: msg.Content, - Channel: origChannel, - ChatID: msg.ChatID, - PeerKind: origPeerKind, - LocalKey: origLocalKey, - UserID: announceUserID, - RunID: fmt.Sprintf("teammate-%s-%s", msg.Metadata["from_agent"], msg.Metadata["to_agent"]), - Stream: false, + SessionKey: sessionKey, + Message: msg.Content, + Channel: origChannel, + ChannelType: origChannelType, + ChatID: msg.ChatID, + PeerKind: origPeerKind, + LocalKey: origLocalKey, + UserID: announceUserID, + RunID: fmt.Sprintf("teammate-%s-%s", msg.Metadata["from_agent"], msg.Metadata["to_agent"]), + Stream: false, }) go func(origCh, chatID, senderID string, meta map[string]string) { @@ -1005,3 +1014,12 @@ func buildAnnounceOutMeta(localKey string) map[string]string { } return meta } + +// resolveChannelType returns the platform type for a channel instance name. +// Returns empty string if channelMgr is nil or channel name is empty. +func resolveChannelType(channelMgr *channels.Manager, name string) string { + if channelMgr == nil || name == "" { + return "" + } + return channelMgr.ChannelTypeForName(name) +} diff --git a/cmd/gateway_cron.go b/cmd/gateway_cron.go index e2246886..feeab4a3 100644 --- a/cmd/gateway_cron.go +++ b/cmd/gateway_cron.go @@ -3,9 +3,11 @@ package cmd import ( "context" "fmt" + "strings" "github.com/nextlevelbuilder/goclaw/internal/agent" "github.com/nextlevelbuilder/goclaw/internal/bus" + "github.com/nextlevelbuilder/goclaw/internal/channels" "github.com/nextlevelbuilder/goclaw/internal/config" "github.com/nextlevelbuilder/goclaw/internal/scheduler" "github.com/nextlevelbuilder/goclaw/internal/sessions" @@ -15,7 +17,7 @@ import ( // makeCronJobHandler creates a cron job handler that routes through the scheduler's cron lane. // This ensures per-session concurrency control (same job can't run concurrently) // and integration with /stop, /stopall commands. -func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg *config.Config) func(job *store.CronJob) (*store.CronJobResult, error) { +func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg *config.Config, channelMgr *channels.Manager) func(job *store.CronJob) (*store.CronJobResult, error) { return func(job *store.CronJob) (*store.CronJobResult, error) { agentID := job.AgentID if agentID == "" { @@ -30,17 +32,26 @@ func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg channel = "cron" } + // Infer peer kind from the stored session metadata (group chats need it + // so that tools like message can route correctly via group APIs). + peerKind := resolveCronPeerKind(job) + + // Resolve channel type for system prompt context. + channelType := resolveChannelType(channelMgr, channel) + // Schedule through cron lane — scheduler handles agent resolution and concurrency outCh := sched.Schedule(context.Background(), scheduler.LaneCron, agent.RunRequest{ - SessionKey: sessionKey, - Message: job.Payload.Message, - Channel: channel, - ChatID: job.Payload.To, - UserID: job.UserID, - RunID: fmt.Sprintf("cron:%s", job.ID), - Stream: false, - TraceName: fmt.Sprintf("Cron [%s] - %s", job.Name, agentID), - TraceTags: []string{"cron"}, + SessionKey: sessionKey, + Message: job.Payload.Message, + Channel: channel, + ChannelType: channelType, + ChatID: job.Payload.To, + PeerKind: peerKind, + UserID: job.UserID, + RunID: fmt.Sprintf("cron:%s", job.ID), + Stream: false, + TraceName: fmt.Sprintf("Cron [%s] - %s", job.Name, agentID), + TraceTags: []string{"cron"}, }) // Block until the scheduled run completes @@ -54,11 +65,15 @@ func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg // If job wants delivery to a channel, send the cron reminder message // directly (not the agent response) so it appears as a bot notification. if job.Payload.Deliver && job.Payload.Channel != "" && job.Payload.To != "" { - msgBus.PublishOutbound(bus.OutboundMessage{ + outMsg := bus.OutboundMessage{ Channel: job.Payload.Channel, ChatID: job.Payload.To, Content: job.Payload.Message, - }) + } + if peerKind == "group" { + outMsg.Metadata = map[string]string{"group_id": job.Payload.To} + } + msgBus.PublishOutbound(outMsg) } cronResult := &store.CronJobResult{ @@ -72,3 +87,12 @@ func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg return cronResult, nil } } + +// resolveCronPeerKind infers peer kind from the cron job's user ID. +// Group cron jobs have userID prefixed with "group:" (set during job creation). +func resolveCronPeerKind(job *store.CronJob) string { + if strings.HasPrefix(job.UserID, "group:") { + return "group" + } + return "" +} diff --git a/internal/agent/loop.go b/internal/agent/loop.go index a9721cb6..4a705b93 100644 --- a/internal/agent/loop.go +++ b/internal/agent/loop.go @@ -197,6 +197,10 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error) if l.builtinToolSettings != nil { ctx = tools.WithBuiltinToolSettings(ctx, l.builtinToolSettings) } + // Inject channel type into context for tools (e.g. message tool needs it for Zalo group routing) + if req.ChannelType != "" { + ctx = tools.WithToolChannelType(ctx, req.ChannelType) + } // Per-user workspace isolation. // Workspace path comes from user_agent_profiles (includes channel segment @@ -295,7 +299,7 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error) // buildMessages resolves context files once and also detects BOOTSTRAP.md presence // (hadBootstrap) — no extra DB roundtrip needed for bootstrap detection. - messages, hadBootstrap := l.buildMessages(ctx, history, summary, req.Message, req.ExtraSystemPrompt, req.SessionKey, req.Channel, req.PeerKind, req.UserID, req.HistoryLimit, req.SkillFilter) + messages, hadBootstrap := l.buildMessages(ctx, history, summary, req.Message, req.ExtraSystemPrompt, req.SessionKey, req.Channel, req.ChannelType, req.PeerKind, req.UserID, req.HistoryLimit, req.SkillFilter) // 1b. Determine image routing strategy. // If read_image tool has a dedicated vision provider, images are NOT attached inline diff --git a/internal/agent/loop_history.go b/internal/agent/loop_history.go index 3dcc35fa..1399609b 100644 --- a/internal/agent/loop_history.go +++ b/internal/agent/loop_history.go @@ -32,7 +32,7 @@ func (l *Loop) filteredToolNames() []string { // buildMessages constructs the full message list for an LLM request. // Returns the messages and whether BOOTSTRAP.md was present in context files // (used by the caller for auto-cleanup without an extra DB roundtrip). -func (l *Loop) buildMessages(ctx context.Context, history []providers.Message, summary, userMessage, extraSystemPrompt, sessionKey, channel, peerKind, userID string, historyLimit int, skillFilter []string) ([]providers.Message, bool) { +func (l *Loop) buildMessages(ctx context.Context, history []providers.Message, summary, userMessage, extraSystemPrompt, sessionKey, channel, channelType, peerKind, userID string, historyLimit int, skillFilter []string) ([]providers.Message, bool) { var messages []providers.Message // Build full system prompt using the new builder (matching TS buildAgentSystemPrompt) @@ -84,6 +84,7 @@ func (l *Loop) buildMessages(ctx context.Context, history []providers.Message, s Model: l.model, Workspace: promptWorkspace, Channel: channel, + ChannelType: channelType, PeerKind: peerKind, OwnerIDs: l.ownerIDs, Mode: mode, diff --git a/internal/agent/loop_types.go b/internal/agent/loop_types.go index 0eca1bc4..b1dfc49c 100644 --- a/internal/agent/loop_types.go +++ b/internal/agent/loop_types.go @@ -272,7 +272,8 @@ type RunRequest struct { Message string // user message Media []bus.MediaFile // local media files with MIME types ForwardMedia []bus.MediaFile // media files to forward to output (from delegation results) - Channel string // source channel + Channel string // source channel instance name (e.g. "my-telegram-bot") + ChannelType string // platform type (e.g. "zalo_personal", "telegram") — for system prompt context ChatID string // source chat ID PeerKind string // "direct" or "group" (for session key building and tool context) RunID string // unique run identifier diff --git a/internal/agent/systemprompt.go b/internal/agent/systemprompt.go index a74fbdc8..5700eed7 100644 --- a/internal/agent/systemprompt.go +++ b/internal/agent/systemprompt.go @@ -23,7 +23,8 @@ type SystemPromptConfig struct { AgentID string Model string Workspace string - Channel string // runtime channel (telegram, discord, etc.) + Channel string // runtime channel instance name (e.g. "my-telegram-bot") + ChannelType string // platform type (e.g. "zalo_personal", "telegram") PeerKind string // "direct" or "group" OwnerIDs []string // owner sender IDs Mode PromptMode // full or minimal @@ -84,13 +85,17 @@ func BuildSystemPrompt(cfg SystemPromptConfig) string { isMinimal := cfg.Mode == PromptMinimal var lines []string - // 1. Identity — channel-aware context - if cfg.Channel != "" { + // 1. Identity — channel-aware context (use ChannelType for clarity, fallback to Channel) + channelLabel := cfg.ChannelType + if channelLabel == "" { + channelLabel = cfg.Channel + } + if channelLabel != "" { chatType := "a direct chat" if cfg.PeerKind == "group" { chatType = "a group chat" } - lines = append(lines, fmt.Sprintf("You are a personal assistant running in %s (%s).", cfg.Channel, chatType)) + lines = append(lines, fmt.Sprintf("You are a personal assistant running in %s (%s).", channelLabel, chatType)) lines = append(lines, "") } diff --git a/internal/channels/channel.go b/internal/channels/channel.go index 7c429b1e..b4cc99d5 100644 --- a/internal/channels/channel.go +++ b/internal/channels/channel.go @@ -50,9 +50,13 @@ const ( // Channel defines the interface that all channel implementations must satisfy. type Channel interface { - // Name returns the channel identifier (e.g., "telegram", "discord", "slack"). + // Name returns the channel instance name (e.g., "telegram", "discord", "slack"). Name() string + // Type returns the platform type (e.g., "telegram", "zalo_personal"). + // For config-based channels this equals Name(); for DB instances it may differ. + Type() string + // Start begins listening for messages. Should be non-blocking after setup. Start(ctx context.Context) error @@ -119,11 +123,12 @@ type ReactionChannel interface { // BaseChannel provides shared functionality for all channel implementations. // Channel implementations should embed this struct. type BaseChannel struct { - name string - bus *bus.MessageBus - running bool - allowList []string - agentID string // for DB instances: routes to specific agent (empty = use resolveAgentRoute) + name string + channelType string // platform type; defaults to name if unset + bus *bus.MessageBus + running bool + allowList []string + agentID string // for DB instances: routes to specific agent (empty = use resolveAgentRoute) } // NewBaseChannel creates a new BaseChannel with the given parameters. @@ -135,12 +140,23 @@ func NewBaseChannel(name string, msgBus *bus.MessageBus, allowList []string) *Ba } } -// Name returns the channel name. +// Name returns the channel instance name. func (c *BaseChannel) Name() string { return c.name } +// Type returns the platform type. Falls back to name if unset (config-based channels). +func (c *BaseChannel) Type() string { + if c.channelType != "" { + return c.channelType + } + return c.name +} + // SetName overrides the channel name (used by InstanceLoader for DB instances). func (c *BaseChannel) SetName(name string) { c.name = name } +// SetType sets the platform type (used by InstanceLoader for DB instances). +func (c *BaseChannel) SetType(t string) { c.channelType = t } + // AgentID returns the explicit agent ID for this channel (empty = use resolveAgentRoute). func (c *BaseChannel) AgentID() string { return c.agentID } diff --git a/internal/channels/instance_loader.go b/internal/channels/instance_loader.go index 9a69b247..cd90e75a 100644 --- a/internal/channels/instance_loader.go +++ b/internal/channels/instance_loader.go @@ -213,7 +213,10 @@ func (l *InstanceLoader) loadInstance(ctx context.Context, inst store.ChannelIns } base.SetAgentID(ag.AgentKey) } - + // Set the platform type on the channel so Manager.ChannelTypeForName can read it. + if base, ok := ch.(interface{ SetType(string) }); ok { + base.SetType(inst.ChannelType) + } l.manager.RegisterChannel(inst.Name, ch) l.loaded[inst.Name] = struct{}{} diff --git a/internal/channels/manager.go b/internal/channels/manager.go index 5506b8b0..91187eaa 100644 --- a/internal/channels/manager.go +++ b/internal/channels/manager.go @@ -210,6 +210,17 @@ func (m *Manager) RegisterChannel(name string, channel Channel) { m.channels[name] = channel } +// ChannelTypeForName returns the platform type for a channel instance name. +// Reads directly from the Channel.Type() method — no separate map needed. +func (m *Manager) ChannelTypeForName(name string) string { + m.mu.RLock() + defer m.mu.RUnlock() + if ch, ok := m.channels[name]; ok { + return ch.Type() + } + return "" +} + // UnregisterChannel removes a channel from the manager. func (m *Manager) UnregisterChannel(name string) { m.mu.Lock() diff --git a/internal/channels/zalo/personal/channel.go b/internal/channels/zalo/personal/channel.go index 312f6c5e..1fafb4af 100644 --- a/internal/channels/zalo/personal/channel.go +++ b/internal/channels/zalo/personal/channel.go @@ -45,6 +45,8 @@ type Channel struct { // Pre-loaded credentials (from DB or from file/QR as fallback). preloadedCreds *protocol.Credentials + groupHistory *channels.PendingHistory + historyLimit int requireMention bool stopCh chan struct{} stopOnce sync.Once @@ -67,10 +69,17 @@ func New(cfg config.ZaloPersonalConfig, msgBus *bus.MessageBus, pairingSvc store requireMention = *cfg.RequireMention } + historyLimit := cfg.HistoryLimit + if historyLimit == 0 { + historyLimit = channels.DefaultGroupHistoryLimit + } + return &Channel{ BaseChannel: base, config: cfg, pairingService: pairingSvc, + groupHistory: channels.NewPendingHistory(), + historyLimit: historyLimit, requireMention: requireMention, stopCh: make(chan struct{}), }, nil @@ -156,9 +165,12 @@ func (c *Channel) Send(ctx context.Context, msg bus.OutboundMessage) error { } threadType := protocol.ThreadTypeUser - if msg.Metadata != nil { + if _, isGroup := c.approvedGroups.Load(msg.ChatID); isGroup { + threadType = protocol.ThreadTypeGroup + } else if msg.Metadata != nil { if _, ok := msg.Metadata["group_id"]; ok { threadType = protocol.ThreadTypeGroup + c.approvedGroups.Store(msg.ChatID, true) } } @@ -371,16 +383,47 @@ func (c *Channel) handleGroupMessage(msg protocol.GroupMessage) { return } - if !c.checkGroupPolicy(senderID, threadID, msg.Data.Mentions) { + // Step 1: enforce access policy (allowlist/pairing). Hard reject — don't record history. + if !c.checkGroupPolicy(senderID, threadID) { return } + senderName := msg.Data.DName + if senderName == "" { + senderName = senderID + } + + // Step 2: @mention gating — record non-mentioned messages in history and return. + if c.requireMention { + wasMentioned := c.checkBotMentioned(msg.Data.Mentions) + if !wasMentioned { + c.groupHistory.Record(threadID, channels.HistoryEntry{ + Sender: senderName, + Body: content, + Timestamp: time.Now(), + MessageID: msg.Data.MsgID, + }, c.historyLimit) + slog.Debug("zalo_personal group message recorded (no mention)", + "group_id", threadID, + "sender", senderName, + ) + return + } + } + slog.Debug("zalo_personal group message received", "sender", senderID, "group", threadID, "preview", channels.Truncate(content, 50), ) + // Step 3: flush pending history + annotate current message with sender name. + annotated := fmt.Sprintf("[From: %s]\n%s", senderName, content) + finalContent := annotated + if c.historyLimit > 0 { + finalContent = c.groupHistory.BuildContext(threadID, annotated, c.historyLimit) + } + c.startTyping(threadID, protocol.ThreadTypeGroup) metadata := map[string]string{ @@ -388,7 +431,7 @@ func (c *Channel) handleGroupMessage(msg protocol.GroupMessage) { "platform": "zalo_personal", "group_id": threadID, } - c.HandleMessage(senderID, threadID, content, media, metadata, "group") + c.HandleMessage(senderID, threadID, finalContent, media, metadata, "group") } // startTyping starts a typing indicator with keepalive for the given thread. diff --git a/internal/channels/zalo/personal/factory.go b/internal/channels/zalo/personal/factory.go index 57f51139..194d545f 100644 --- a/internal/channels/zalo/personal/factory.go +++ b/internal/channels/zalo/personal/factory.go @@ -24,6 +24,7 @@ type zaloInstanceConfig struct { DMPolicy string `json:"dm_policy,omitempty"` GroupPolicy string `json:"group_policy,omitempty"` RequireMention *bool `json:"require_mention,omitempty"` + HistoryLimit int `json:"history_limit,omitempty"` AllowFrom []string `json:"allow_from,omitempty"` BlockReply *bool `json:"block_reply,omitempty"` } @@ -59,6 +60,7 @@ func Factory(name string, creds json.RawMessage, cfg json.RawMessage, DMPolicy: ic.DMPolicy, GroupPolicy: ic.GroupPolicy, RequireMention: ic.RequireMention, + HistoryLimit: ic.HistoryLimit, BlockReply: ic.BlockReply, } diff --git a/internal/channels/zalo/personal/policy.go b/internal/channels/zalo/personal/policy.go index 037b90c2..f4f645bb 100644 --- a/internal/channels/zalo/personal/policy.go +++ b/internal/channels/zalo/personal/policy.go @@ -89,8 +89,9 @@ func (c *Channel) sendPairingReply(senderID, chatID string) { } } -// checkGroupPolicy enforces group policy and @mention gating. -func (c *Channel) checkGroupPolicy(senderID, groupID string, mentions []*protocol.TMention) bool { +// checkGroupPolicy enforces group access policy (allowlist/pairing). +// Returns false if the group is blocked by policy; does NOT check @mention gating. +func (c *Channel) checkGroupPolicy(senderID, groupID string) bool { groupPolicy := c.config.GroupPolicy if groupPolicy == "" { groupPolicy = "allowlist" @@ -123,25 +124,18 @@ func (c *Channel) checkGroupPolicy(senderID, groupID string, mentions []*protoco } } - // @mention gating: only process group messages that @mention the bot. - if c.requireMention { - sess := c.session() - botUID := "" - if sess != nil { - botUID = sess.UID - } - if !isBotMentioned(botUID, mentions) { - slog.Debug("zalo_personal group message skipped: not mentioned", - "group_id", groupID, - "sender_id", senderID, - ) - return false - } - } - return true } +// checkBotMentioned reports whether the bot is @mentioned in the message. +func (c *Channel) checkBotMentioned(mentions []*protocol.TMention) bool { + sess := c.session() + if sess == nil { + return false + } + return isBotMentioned(sess.UID, mentions) +} + // isBotMentioned checks if the bot's UID is @mentioned in the message. // Filters out @all mentions (Type=1, UID="-1") — only targeted @bot counts. func isBotMentioned(botUID string, mentions []*protocol.TMention) bool { diff --git a/internal/config/config_channels.go b/internal/config/config_channels.go index 7b64a939..16ac4d85 100644 --- a/internal/config/config_channels.go +++ b/internal/config/config_channels.go @@ -132,6 +132,7 @@ type ZaloPersonalConfig struct { DMPolicy string `json:"dm_policy,omitempty"` // "pairing" (default), "allowlist", "open", "disabled" GroupPolicy string `json:"group_policy,omitempty"` // "open" (default), "allowlist", "disabled" RequireMention *bool `json:"require_mention,omitempty"` // require @bot mention in groups (default true) + HistoryLimit int `json:"history_limit,omitempty"` // max pending group messages for context (default 50, 0=disabled) CredentialsPath string `json:"credentials_path,omitempty"` // path to saved cookies JSON BlockReply *bool `json:"block_reply,omitempty"` // override gateway block_reply (nil = inherit) } diff --git a/internal/tools/context_keys.go b/internal/tools/context_keys.go index d8f0d7b7..319aae5e 100644 --- a/internal/tools/context_keys.go +++ b/internal/tools/context_keys.go @@ -14,9 +14,10 @@ import ( type toolContextKey string const ( - ctxChannel toolContextKey = "tool_channel" - ctxChatID toolContextKey = "tool_chat_id" - ctxPeerKind toolContextKey = "tool_peer_kind" + ctxChannel toolContextKey = "tool_channel" + ctxChannelType toolContextKey = "tool_channel_type" + ctxChatID toolContextKey = "tool_chat_id" + ctxPeerKind toolContextKey = "tool_peer_kind" ctxLocalKey toolContextKey = "tool_local_key" // composite key with topic/thread suffix for routing ctxSandboxKey toolContextKey = "tool_sandbox_key" ctxAsyncCB toolContextKey = "tool_async_cb" @@ -34,6 +35,15 @@ func ToolChannelFromCtx(ctx context.Context) string { return v } +func WithToolChannelType(ctx context.Context, channelType string) context.Context { + return context.WithValue(ctx, ctxChannelType, channelType) +} + +func ToolChannelTypeFromCtx(ctx context.Context) string { + v, _ := ctx.Value(ctxChannelType).(string) + return v +} + func WithToolChatID(ctx context.Context, chatID string) context.Context { return context.WithValue(ctx, ctxChatID, chatID) } diff --git a/internal/tools/cron.go b/internal/tools/cron.go index 7cbac075..9b309320 100644 --- a/internal/tools/cron.go +++ b/internal/tools/cron.go @@ -45,7 +45,7 @@ JOB SCHEMA (for add action): "schedule": { ... }, // Required: when to run "message": "string", // Required: what message to send to the agent "deliver": true|false, // Optional: deliver result to channel (default false) - "channel": "telegram", // Optional: target channel for delivery + "channel": "channel-name", // Optional: target channel for delivery (auto-filled from context) "to": "chat-id", // Optional: target chat/recipient ID "agentId": "agent-uuid", // Optional: which agent handles the job (default: current) "deleteAfterRun": true // Optional: auto-delete after execution (default true for "at" schedule) diff --git a/internal/tools/message.go b/internal/tools/message.go index 8cda910d..94455bc4 100644 --- a/internal/tools/message.go +++ b/internal/tools/message.go @@ -9,6 +9,7 @@ import ( "strings" "github.com/nextlevelbuilder/goclaw/internal/bus" + "github.com/nextlevelbuilder/goclaw/internal/store" ) // MessageTool allows the agent to proactively send messages to channels. @@ -24,7 +25,7 @@ func (t *MessageTool) SetMessageBus(b *bus.MessageBus) { t.msgBus = b } func (t *MessageTool) Name() string { return "message" } func (t *MessageTool) Description() string { - return "Send a message to a channel (Telegram, Discord, etc.) or the current chat." + return "Send a message to a channel (Telegram, Discord, Slack, Zalo, Feishu/Lark, WhatsApp, etc.) or the current chat. Channel and target are auto-filled from context." } func (t *MessageTool) Parameters() map[string]interface{} { @@ -85,22 +86,37 @@ func (t *MessageTool) Execute(ctx context.Context, args map[string]interface{}) return t.sendMedia(ctx, channel, target, filePath) } - // Prefer direct channel sender (channels.Manager.SendToChannel) - if t.sender != nil { + // Prefer direct channel sender for immediate delivery. + // For group chats, fall through to message bus which supports metadata. + if t.sender != nil && !isGroupContext(ctx) { if err := t.sender(ctx, channel, target, message); err != nil { return ErrorResult(fmt.Sprintf("failed to send message: %v", err)) } return SilentResult(fmt.Sprintf(`{"status":"sent","channel":"%s","target":"%s"}`, channel, target)) } - // Fallback: publish via message bus outbound queue + // Publish via message bus outbound queue. + // Group messages include metadata so channel implementations (e.g. Zalo) + // can distinguish group sends from DMs. if t.msgBus != nil { - t.msgBus.PublishOutbound(bus.OutboundMessage{ + outMsg := bus.OutboundMessage{ Channel: channel, ChatID: target, Content: message, - }) - return SilentResult(fmt.Sprintf(`{"status":"queued","channel":"%s","target":"%s"}`, channel, target)) + } + if isGroupContext(ctx) { + outMsg.Metadata = map[string]string{"group_id": target} + } + t.msgBus.PublishOutbound(outMsg) + return SilentResult(fmt.Sprintf(`{"status":"sent","channel":"%s","target":"%s"}`, channel, target)) + } + + // Last resort: direct sender without group metadata. + if t.sender != nil { + if err := t.sender(ctx, channel, target, message); err != nil { + return ErrorResult(fmt.Sprintf("failed to send message: %v", err)) + } + return SilentResult(fmt.Sprintf(`{"status":"sent","channel":"%s","target":"%s"}`, channel, target)) } return ErrorResult("no channel sender or message bus available") @@ -117,8 +133,7 @@ func (t *MessageTool) sendMedia(ctx context.Context, channel, target, filePath s // Build metadata for group routing (Zalo needs group_id to choose group API). var meta map[string]string - peerKind := ToolPeerKindFromCtx(ctx) - if peerKind == "group" { + if isGroupContext(ctx) { meta = map[string]string{"group_id": target} } @@ -137,6 +152,12 @@ func (t *MessageTool) sendMedia(ctx context.Context, channel, target, filePath s return SilentResult(string(out)) } +// isGroupContext returns true if the current context indicates a group conversation. +func isGroupContext(ctx context.Context) bool { + return ToolPeerKindFromCtx(ctx) == "group" || + strings.HasPrefix(store.UserIDFromContext(ctx), "group:") +} + // parseMediaPath extracts a file path from a "MEDIA:/path/to/file" string. // Only allows absolute paths within os.TempDir() to prevent path traversal. func parseMediaPath(s string) (string, bool) {