diff --git a/cmd/gateway.go b/cmd/gateway.go index 9f61bac3..9d24ec2b 100644 --- a/cmd/gateway.go +++ b/cmd/gateway.go @@ -737,6 +737,20 @@ func runGateway() { } } + // Register create_forum_topic tool (lazy bot resolution via channel manager). + toolsReg.Register(tools.NewCreateForumTopicTool(func() tools.ForumTopicCreator { + for _, name := range channelMgr.GetEnabledChannels() { + ch, ok := channelMgr.GetChannel(name) + if !ok { + continue + } + if fc, ok := ch.(tools.ForumTopicCreator); ok { + return fc + } + } + return nil + })) + // Register channels RPC methods (after channelMgr is initialized with all channels) methods.NewChannelsMethods(channelMgr).Register(server.Router()) diff --git a/cmd/gateway_consumer.go b/cmd/gateway_consumer.go index a4dbb88e..27d73be2 100644 --- a/cmd/gateway_consumer.go +++ b/cmd/gateway_consumer.go @@ -88,6 +88,15 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents } } + // DM thread: override session key to isolate per-thread history in private chats. + if msg.Metadata["dm_thread_id"] != "" && peerKind == string(sessions.PeerDirect) { + var threadID int + fmt.Sscanf(msg.Metadata["dm_thread_id"], "%d", &threadID) + if threadID > 0 { + sessionKey = sessions.BuildDMThreadSessionKey(agentID, msg.Channel, msg.ChatID, threadID) + } + } + // Group-scoped UserID: treat the group as a single "virtual user" for // context files, memory, traces, and seeding. Individual senderID is // preserved in the InboundMessage for pairing/dedup/mention gate. @@ -152,6 +161,20 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents "- Address the group naturally. If the history shows a multi-person conversation, consider the full context before answering." } + // Append per-topic system prompt (from group/topic config hierarchy). + if tsp := msg.Metadata["topic_system_prompt"]; tsp != "" { + if extraPrompt != "" { + extraPrompt += "\n\n" + } + extraPrompt += tsp + } + + // Per-topic skill filter override (from group/topic config hierarchy). + var skillFilter []string + if ts := msg.Metadata["topic_skills"]; ts != "" { + skillFilter = strings.Split(ts, ",") + } + // Delegation announces carry media as ForwardMedia (not deleted, forwarded to output). // User-uploaded media goes in Media (loaded as images for LLM, then deleted). var reqMedia, fwdMedia []string @@ -176,6 +199,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents Stream: enableStream, HistoryLimit: msg.HistoryLimit, ExtraSystemPrompt: extraPrompt, + SkillFilter: skillFilter, }, scheduler.ScheduleOpts{ MaxConcurrent: maxConcurrent, }) @@ -629,6 +653,13 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents sessionKey = sessions.BuildGroupTopicSessionKey(agentID, msg.Channel, msg.ChatID, topicID) } } + if msg.Metadata["dm_thread_id"] != "" && peerKind == string(sessions.PeerDirect) { + var threadID int + fmt.Sscanf(msg.Metadata["dm_thread_id"], "%d", &threadID) + if threadID > 0 { + sessionKey = sessions.BuildDMThreadSessionKey(agentID, msg.Channel, msg.ChatID, threadID) + } + } var cancelled bool if cmd == "stopall" { diff --git a/internal/agent/loop.go b/internal/agent/loop.go index 9a234266..4fe3c801 100644 --- a/internal/agent/loop.go +++ b/internal/agent/loop.go @@ -244,8 +244,9 @@ type RunRequest struct { UserID string // external user ID (TEXT, free-form) for multi-tenant scoping SenderID string // original individual sender ID (preserved in group chats for permission checks) Stream bool // whether to stream response chunks - ExtraSystemPrompt string // optional: injected into system prompt (skills, subagent context, etc.) - HistoryLimit int // max user turns to keep in context (0=unlimited, from channel config) + ExtraSystemPrompt string // optional: injected into system prompt (skills, subagent context, etc.) + SkillFilter []string // per-request skill override: nil=use agent default, []=no skills, ["x","y"]=whitelist + HistoryLimit int // max user turns to keep in context (0=unlimited, from channel config) ParentTraceID uuid.UUID // if set, reuse parent trace instead of creating new (announce runs) ParentRootSpanID uuid.UUID // if set, nest announce agent span under this parent span TraceName string // override trace name (default: "chat ") @@ -495,7 +496,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.UserID, req.HistoryLimit) + messages, hadBootstrap := l.buildMessages(ctx, history, summary, req.Message, req.ExtraSystemPrompt, req.SessionKey, req.Channel, req.UserID, req.HistoryLimit, req.SkillFilter) // 2. Attach vision images to the current user message (last in messages slice). // Images are only attached to the live request, NOT persisted in session history. diff --git a/internal/agent/loop_history.go b/internal/agent/loop_history.go index d10fc5b9..aaf17ec4 100644 --- a/internal/agent/loop_history.go +++ b/internal/agent/loop_history.go @@ -16,7 +16,7 @@ import ( // 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, userID string, historyLimit int) ([]providers.Message, bool) { +func (l *Loop) buildMessages(ctx context.Context, history []providers.Message, summary, userMessage, extraSystemPrompt, sessionKey, channel, userID string, historyLimit int, skillFilter []string) ([]providers.Message, bool) { var messages []providers.Message // Build full system prompt using the new builder (matching TS buildAgentSystemPrompt) @@ -57,7 +57,7 @@ func (l *Loop) buildMessages(ctx context.Context, history []providers.Message, s OwnerIDs: l.ownerIDs, Mode: mode, ToolNames: l.tools.List(), - SkillsSummary: l.resolveSkillsSummary(), + SkillsSummary: l.resolveSkillsSummary(skillFilter), HasMemory: l.hasMemory, HasSpawn: l.tools != nil && hasSpawn, HasSkillSearch: hasSkillSearch, @@ -142,12 +142,18 @@ const ( // Returns (summary XML, useInline) — useInline=true means skills are inlined and // the system prompt should use TS-style "scan " instructions // instead of "use skill_search". -func (l *Loop) resolveSkillsSummary() string { +func (l *Loop) resolveSkillsSummary(skillFilter []string) string { if l.skillsLoader == nil { return "" } - filtered := l.skillsLoader.FilterSkills(l.skillAllowList) + // Per-request skill filter overrides agent-level allowList. + allowList := l.skillAllowList + if skillFilter != nil { + allowList = skillFilter + } + + filtered := l.skillsLoader.FilterSkills(allowList) if len(filtered) == 0 { return "" } @@ -161,7 +167,7 @@ func (l *Loop) resolveSkillsSummary() string { if len(filtered) <= skillInlineMaxCount && estimatedTokens <= skillInlineMaxTokens { // Inline mode: build full XML summary - return l.skillsLoader.BuildSummary(l.skillAllowList) + return l.skillsLoader.BuildSummary(allowList) } // Search mode: no XML in prompt, agent uses skill_search tool diff --git a/internal/channels/telegram/channel.go b/internal/channels/telegram/channel.go index 2ff33a18..4c05ea07 100644 --- a/internal/channels/telegram/channel.go +++ b/internal/channels/telegram/channel.go @@ -229,10 +229,33 @@ func parseRawChatID(key string) (int64, error) { raw := key if idx := strings.Index(key, ":topic:"); idx > 0 { raw = key[:idx] + } else if idx := strings.Index(key, ":thread:"); idx > 0 { + raw = key[:idx] } return parseChatID(raw) } +// CreateForumTopic creates a new forum topic in a supergroup. +// Implements tools.ForumTopicCreator interface. +func (c *Channel) CreateForumTopic(ctx context.Context, chatID int64, name string, iconColor int, iconEmojiID string) (int, string, error) { + params := &telego.CreateForumTopicParams{ + ChatID: telego.ChatID{ID: chatID}, + Name: name, + } + if iconColor > 0 { + params.IconColor = iconColor + } + if iconEmojiID != "" { + params.IconCustomEmojiID = iconEmojiID + } + + topic, err := c.bot.CreateForumTopic(ctx, params) + if err != nil { + return 0, "", fmt.Errorf("telegram API: %w", err) + } + return topic.MessageThreadID, topic.Name, nil +} + // telegramGeneralTopicID is the fixed topic ID for the "General" topic in forum supergroups. // TS ref: TELEGRAM_GENERAL_TOPIC_ID in src/telegram/bot/helpers.ts:12. const telegramGeneralTopicID = 1 diff --git a/internal/channels/telegram/commands_pairing.go b/internal/channels/telegram/commands_pairing.go index fed331dd..72525b36 100644 --- a/internal/channels/telegram/commands_pairing.go +++ b/internal/channels/telegram/commands_pairing.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "log/slog" + "strings" "time" "github.com/mymmrac/telego" @@ -54,14 +55,17 @@ func (c *Channel) sendPairingReply(ctx context.Context, chatID int64, userID, us // sendGroupPairingReply generates a pairing code for a group and sends the reply. // Debounces: won't send another reply to the same group within 60 seconds. -func (c *Channel) sendGroupPairingReply(ctx context.Context, chatID int64, chatIDStr, groupSenderID string) { +// messageThreadID should be set for forum groups so the reply lands in the correct topic. +// localKey is the composite key (e.g. "-100123:topic:42") stored as chat_id in the pairing +// request so that the approval notification can be routed to the correct forum topic. +func (c *Channel) sendGroupPairingReply(ctx context.Context, chatID int64, chatIDStr, groupSenderID, localKey string, messageThreadID int) { if lastSent, ok := c.pairingReplySent.Load(chatIDStr); ok { if time.Since(lastSent.(time.Time)) < pairingReplyDebounce { return } } - code, err := c.pairingService.RequestPairing(groupSenderID, c.Name(), chatIDStr, "default") + code, err := c.pairingService.RequestPairing(groupSenderID, c.Name(), localKey, "default") if err != nil { slog.Debug("group pairing request failed", "chat_id", chatIDStr, "error", err) return @@ -72,6 +76,9 @@ func (c *Channel) sendGroupPairingReply(ctx context.Context, chatID int64, chatI code, code, ) msg := tu.Message(tu.ID(chatID), replyText) + if messageThreadID > 0 { + msg.MessageThreadID = messageThreadID + } if _, err := c.bot.SendMessage(ctx, msg); err != nil { slog.Warn("failed to send group pairing reply", "chat_id", chatIDStr, "error", err) } else { @@ -81,8 +88,9 @@ func (c *Channel) sendGroupPairingReply(ctx context.Context, chatID int64, chatI } // SendPairingApproved sends the approval notification to a user. +// chatID may contain a topic suffix (e.g. "-100123:topic:42") for forum groups. func (c *Channel) SendPairingApproved(ctx context.Context, chatID, botName string) error { - id, err := parseChatID(chatID) + id, err := parseRawChatID(chatID) if err != nil { return fmt.Errorf("invalid chat ID: %w", err) } @@ -91,6 +99,22 @@ func (c *Channel) SendPairingApproved(ctx context.Context, chatID, botName strin } msg := tu.Message(tu.ID(id), fmt.Sprintf("✅ %s access approved. Send a message to start chatting.", botName)) + + // Extract thread ID from topic/thread suffix for forum groups. + if idx := strings.Index(chatID, ":topic:"); idx > 0 { + var threadID int + fmt.Sscanf(chatID[idx+7:], "%d", &threadID) + if threadID > 0 { + msg.MessageThreadID = threadID + } + } else if idx := strings.Index(chatID, ":thread:"); idx > 0 { + var threadID int + fmt.Sscanf(chatID[idx+8:], "%d", &threadID) + if threadID > 0 { + msg.MessageThreadID = threadID + } + } + _, err = c.bot.SendMessage(ctx, msg) return err } diff --git a/internal/channels/telegram/handlers.go b/internal/channels/telegram/handlers.go index 381d8e3d..b837ab9f 100644 --- a/internal/channels/telegram/handlers.go +++ b/internal/channels/telegram/handlers.go @@ -69,9 +69,33 @@ func (c *Channel) handleMessage(ctx context.Context, update telego.Update) { } } - // Group policy check (matching TS: groupPolicy ?? "open"). + // DM thread detection: preserve message_thread_id in private chats for session isolation. + // Telegram supports topics/threads in bot DMs. + dmThreadID := 0 + if !isGroup && message.MessageThreadID > 0 { + dmThreadID = message.MessageThreadID + } + + chatID := message.Chat.ID + chatIDStr := fmt.Sprintf("%d", chatID) + + // Resolve per-topic config (matching TS resolveTelegramGroupConfig). + // Merges: global defaults → wildcard group ("*") → specific group → specific topic. + var topicCfg resolvedTopicConfig if isGroup { - groupPolicy := c.config.GroupPolicy + topicCfg = resolveTopicConfig(c.config, chatIDStr, messageThreadID) + } + + // Group policy + enabled check (matching TS: groupPolicy ?? "open"). + if isGroup { + // Per-topic enabled gate: if explicitly disabled, reject. + if !topicCfg.isEnabled() { + slog.Debug("telegram group message rejected: topic disabled", + "chat_id", chatID, "topic_id", messageThreadID) + return + } + + groupPolicy := topicCfg.groupPolicy if groupPolicy == "" { groupPolicy = "open" } @@ -81,9 +105,16 @@ func (c *Channel) handleMessage(ctx context.Context, update telego.Update) { slog.Debug("telegram group message rejected: groups disabled", "chat_id", message.Chat.ID) return case "allowlist": - if !c.IsAllowed(userID) && !c.IsAllowed(senderID) { + allowed := false + for _, a := range topicCfg.allowFrom { + if a == userID || a == senderID { + allowed = true + break + } + } + if !allowed { slog.Debug("telegram group message rejected by allowlist", - "user_id", userID, "username", user.Username, "chat_id", message.Chat.ID, + "user_id", userID, "username", user.Username, "chat_id", chatID, ) return } @@ -131,20 +162,21 @@ func (c *Channel) handleMessage(ctx context.Context, update telego.Update) { } } - chatID := message.Chat.ID - chatIDStr := fmt.Sprintf("%d", chatID) - // Build composite localKey for sync.Map operations. // Forum topics get separate state (placeholders, streams, reactions, history). // TS ref: buildTelegramGroupPeerId() in src/telegram/bot/helpers.ts. localKey := chatIDStr if isForum && messageThreadID > 0 { localKey = fmt.Sprintf("%s:topic:%d", chatIDStr, messageThreadID) + } else if dmThreadID > 0 { + localKey = fmt.Sprintf("%s:thread:%d", chatIDStr, dmThreadID) } // Store thread ID for streaming/send use (looked up by localKey later). if messageThreadID > 0 { c.threadIDs.Store(localKey, messageThreadID) + } else if dmThreadID > 0 { + c.threadIDs.Store(localKey, dmThreadID) } // Extract text content @@ -245,7 +277,7 @@ func (c *Channel) handleMessage(ctx context.Context, update telego.Update) { // --- Group mention gating (matching TS mentionGate logic) --- // Also check implicit mention via reply-to-bot - if isGroup && c.requireMention { + if isGroup && topicCfg.effectiveRequireMention(c.requireMention) { botUsername := c.bot.Username() wasMentioned := c.detectMention(message, botUsername) @@ -278,13 +310,13 @@ func (c *Channel) handleMessage(ctx context.Context, update telego.Update) { } // --- Group pairing gate (only reached when bot is mentioned) --- - if isGroup && c.config.GroupPolicy == "pairing" && c.pairingService != nil { + if isGroup && topicCfg.groupPolicy == "pairing" && c.pairingService != nil { if _, cached := c.approvedGroups.Load(chatIDStr); !cached { groupSenderID := fmt.Sprintf("group:%d", chatID) if c.pairingService.IsPaired(groupSenderID, c.Name()) { c.approvedGroups.Store(chatIDStr, true) } else { - c.sendGroupPairingReply(ctx, chatID, chatIDStr, groupSenderID) + c.sendGroupPairingReply(ctx, chatID, chatIDStr, groupSenderID, localKey, messageThreadID) return } } @@ -346,9 +378,8 @@ func (c *Channel) handleMessage(ctx context.Context, update telego.Update) { // instead the response will be sent as a reply to the sender's message. if !isGroup { thinkMsg := tu.Message(chatIDObj, "Thinking...") - sendThreadID := resolveThreadIDForSend(messageThreadID) - if sendThreadID > 0 { - thinkMsg.MessageThreadID = sendThreadID + if dmThreadID > 0 { + thinkMsg.MessageThreadID = dmThreadID } pMsg, err := c.bot.SendMessage(ctx, thinkMsg) if err == nil { @@ -368,6 +399,16 @@ func (c *Channel) handleMessage(ctx context.Context, update telego.Update) { metadata["is_forum"] = "true" metadata["message_thread_id"] = fmt.Sprintf("%d", messageThreadID) } + if dmThreadID > 0 { + metadata["dm_thread_id"] = fmt.Sprintf("%d", dmThreadID) + metadata["message_thread_id"] = fmt.Sprintf("%d", dmThreadID) + } + if topicCfg.systemPrompt != "" { + metadata["topic_system_prompt"] = topicCfg.systemPrompt + } + if topicCfg.skills != nil { + metadata["topic_skills"] = strings.Join(topicCfg.skills, ",") + } peerKind := "direct" if isGroup { diff --git a/internal/channels/telegram/send.go b/internal/channels/telegram/send.go index 45d53a81..2a7be67e 100644 --- a/internal/channels/telegram/send.go +++ b/internal/channels/telegram/send.go @@ -21,6 +21,7 @@ import ( var ( parseErrRe = regexp.MustCompile(`(?i)can't parse entities|parse entities|find end of the entity`) messageNotModifiedRe = regexp.MustCompile(`(?i)message is not modified`) + threadNotFoundRe = regexp.MustCompile(`(?i)message thread not found`) htmlTagRe = regexp.MustCompile(`<[^>]*>`) ) @@ -108,6 +109,17 @@ func (c *Channel) Send(ctx context.Context, msg bus.OutboundMessage) error { fmt.Sscanf(v, "%d", &threadID) } + // Fallback: extract threadID from localKey suffix (e.g. "-100123:topic:42"). + // This covers cases where metadata is absent, such as pairing approval notifications + // routed via SendToChannel which only has a chatID string. + if threadID == 0 { + if idx := strings.Index(localKey, ":topic:"); idx > 0 { + fmt.Sscanf(localKey[idx+7:], "%d", &threadID) + } else if idx := strings.Index(localKey, ":thread:"); idx > 0 { + fmt.Sscanf(localKey[idx+8:], "%d", &threadID) + } + } + // Placeholder update (e.g. LLM retry notification): edit the placeholder // but keep it alive for the final response. Don't stop typing or cleanup. if msg.Metadata["placeholder_update"] == "true" { @@ -280,6 +292,12 @@ func (c *Channel) sendHTML(ctx context.Context, chatID int64, html string, reply tgMsg.Text = stripHTML(tgMsg.Text) _, err = c.bot.SendMessage(ctx, tgMsg) } + // TS ref: withTelegramThreadFallback — retry without thread ID when topic is deleted. + if err != nil && tgMsg.MessageThreadID != 0 && threadNotFoundRe.MatchString(err.Error()) { + slog.Warn("thread not found, retrying without message_thread_id", "thread_id", tgMsg.MessageThreadID) + tgMsg.MessageThreadID = 0 + _, err = c.bot.SendMessage(ctx, tgMsg) + } return err } @@ -317,6 +335,12 @@ func (c *Channel) sendPhoto(ctx context.Context, chatID telego.ChatID, filePath, params.Caption = stripHTML(params.Caption) _, err = c.bot.SendPhoto(ctx, params) } + if err != nil && params.MessageThreadID != 0 && threadNotFoundRe.MatchString(err.Error()) { + slog.Warn("sendPhoto: thread not found, retrying without thread", "thread_id", params.MessageThreadID) + file.Seek(0, 0) + params.MessageThreadID = 0 + _, err = c.bot.SendPhoto(ctx, params) + } return err } @@ -354,6 +378,12 @@ func (c *Channel) sendVideo(ctx context.Context, chatID telego.ChatID, filePath, params.Caption = stripHTML(params.Caption) _, err = c.bot.SendVideo(ctx, params) } + if err != nil && params.MessageThreadID != 0 && threadNotFoundRe.MatchString(err.Error()) { + slog.Warn("sendVideo: thread not found, retrying without thread", "thread_id", params.MessageThreadID) + file.Seek(0, 0) + params.MessageThreadID = 0 + _, err = c.bot.SendVideo(ctx, params) + } return err } @@ -391,6 +421,12 @@ func (c *Channel) sendAudio(ctx context.Context, chatID telego.ChatID, filePath, params.Caption = stripHTML(params.Caption) _, err = c.bot.SendAudio(ctx, params) } + if err != nil && params.MessageThreadID != 0 && threadNotFoundRe.MatchString(err.Error()) { + slog.Warn("sendAudio: thread not found, retrying without thread", "thread_id", params.MessageThreadID) + file.Seek(0, 0) + params.MessageThreadID = 0 + _, err = c.bot.SendAudio(ctx, params) + } return err } @@ -428,6 +464,12 @@ func (c *Channel) sendDocument(ctx context.Context, chatID telego.ChatID, filePa params.Caption = stripHTML(params.Caption) _, err = c.bot.SendDocument(ctx, params) } + if err != nil && params.MessageThreadID != 0 && threadNotFoundRe.MatchString(err.Error()) { + slog.Warn("sendDocument: thread not found, retrying without thread", "thread_id", params.MessageThreadID) + file.Seek(0, 0) + params.MessageThreadID = 0 + _, err = c.bot.SendDocument(ctx, params) + } return err } diff --git a/internal/channels/telegram/stream.go b/internal/channels/telegram/stream.go index 5b8c2239..ea1f4648 100644 --- a/internal/channels/telegram/stream.go +++ b/internal/channels/telegram/stream.go @@ -112,6 +112,12 @@ func (ds *DraftStream) flush(ctx context.Context) error { params.MessageThreadID = sendThreadID } msg, err := ds.bot.SendMessage(ctx, params) + // TS ref: withTelegramThreadFallback — retry without thread ID when topic is deleted. + if err != nil && params.MessageThreadID != 0 && threadNotFoundRe.MatchString(err.Error()) { + slog.Warn("stream: thread not found, retrying without message_thread_id", "thread_id", params.MessageThreadID) + params.MessageThreadID = 0 + msg, err = ds.bot.SendMessage(ctx, params) + } if err != nil { slog.Debug("stream: failed to send initial message", "error", err) return err diff --git a/internal/channels/telegram/topic_config.go b/internal/channels/telegram/topic_config.go new file mode 100644 index 00000000..1673baff --- /dev/null +++ b/internal/channels/telegram/topic_config.go @@ -0,0 +1,124 @@ +package telegram + +import ( + "fmt" + "strings" + + "github.com/nextlevelbuilder/goclaw/internal/config" +) + +// resolvedTopicConfig holds the merged config for a specific group+topic combination. +// Fields are resolved in order: global TelegramConfig → wildcard group ("*") → specific group → specific topic. +// TS ref: resolveTelegramGroupConfig() in src/telegram/bot.ts + resolveTelegramGroupPromptSettings() in group-config-helpers.ts. +type resolvedTopicConfig struct { + groupPolicy string + requireMention *bool + allowFrom []string + enabled *bool + skills []string // nil = inherit, non-nil = override (empty = no skills) + systemPrompt string // concatenated group + topic prompts +} + +// resolveTopicConfig resolves the effective config for a chat/topic by merging layers. +// chatIDStr is the raw Telegram chat ID (e.g., "-100123456"). +// topicID is the forum topic thread ID (0 = not a forum topic). +func resolveTopicConfig(cfg config.TelegramConfig, chatIDStr string, topicID int) resolvedTopicConfig { + result := resolvedTopicConfig{ + groupPolicy: cfg.GroupPolicy, + requireMention: cfg.RequireMention, + allowFrom: cfg.AllowFrom, + } + + if cfg.Groups == nil { + return result + } + + // Layer 1: wildcard group config ("*") — applies to all groups unless overridden. + if wildcard, ok := cfg.Groups["*"]; ok && wildcard != nil { + mergeGroupInto(&result, wildcard) + } + + // Layer 2: specific group config — overrides wildcard. + groupCfg, ok := cfg.Groups[chatIDStr] + if !ok || groupCfg == nil { + return result + } + mergeGroupInto(&result, groupCfg) + + // Layer 3: specific topic config — overrides group. + if topicID > 0 && groupCfg.Topics != nil { + topicIDStr := fmt.Sprintf("%d", topicID) + if topicCfg, ok := groupCfg.Topics[topicIDStr]; ok && topicCfg != nil { + mergeTopicInto(&result, topicCfg, groupCfg) + } + } + + return result +} + +// mergeGroupInto applies non-zero group config values over the current result. +func mergeGroupInto(dst *resolvedTopicConfig, src *config.TelegramGroupConfig) { + if src.GroupPolicy != "" { + dst.groupPolicy = src.GroupPolicy + } + if src.RequireMention != nil { + dst.requireMention = src.RequireMention + } + if len(src.AllowFrom) > 0 { + dst.allowFrom = src.AllowFrom + } + if src.Enabled != nil { + dst.enabled = src.Enabled + } + if src.Skills != nil { + dst.skills = src.Skills + } + if src.SystemPrompt != "" { + dst.systemPrompt = src.SystemPrompt + } +} + +// mergeTopicInto applies topic config values, with special handling for systemPrompt +// (concatenated: group + topic, matching TS resolveTelegramGroupPromptSettings). +func mergeTopicInto(dst *resolvedTopicConfig, src *config.TelegramTopicConfig, groupCfg *config.TelegramGroupConfig) { + if src.GroupPolicy != "" { + dst.groupPolicy = src.GroupPolicy + } + if src.RequireMention != nil { + dst.requireMention = src.RequireMention + } + if len(src.AllowFrom) > 0 { + dst.allowFrom = src.AllowFrom + } + if src.Enabled != nil { + dst.enabled = src.Enabled + } + // Skills: topic overrides group (firstDefined pattern, matching TS). + if src.Skills != nil { + dst.skills = src.Skills + } + // SystemPrompt: concatenate group + topic (both may exist, matching TS). + if src.SystemPrompt != "" { + parts := []string{} + if groupCfg.SystemPrompt != "" { + parts = append(parts, strings.TrimSpace(groupCfg.SystemPrompt)) + } + parts = append(parts, strings.TrimSpace(src.SystemPrompt)) + dst.systemPrompt = strings.Join(parts, "\n\n") + } +} + +// isEnabled returns whether the resolved config allows the bot to operate. +// nil = enabled (default), false = disabled. +func (r *resolvedTopicConfig) isEnabled() bool { + return r.enabled == nil || *r.enabled +} + +// effectiveRequireMention returns the resolved require_mention value. +// Falls back to the provided default if not overridden. +func (r *resolvedTopicConfig) effectiveRequireMention(defaultVal bool) bool { + if r.requireMention != nil { + return *r.requireMention + } + return defaultVal +} diff --git a/internal/channels/telegram/topic_config_test.go b/internal/channels/telegram/topic_config_test.go new file mode 100644 index 00000000..4a167ebb --- /dev/null +++ b/internal/channels/telegram/topic_config_test.go @@ -0,0 +1,261 @@ +package telegram + +import ( + "testing" + + "github.com/nextlevelbuilder/goclaw/internal/config" +) + +func boolPtr(b bool) *bool { return &b } + +func TestResolveTopicConfig_Defaults(t *testing.T) { + cfg := config.TelegramConfig{ + GroupPolicy: "open", + RequireMention: boolPtr(true), + AllowFrom: []string{"user1"}, + } + + result := resolveTopicConfig(cfg, "-100123", 0) + + if result.groupPolicy != "open" { + t.Errorf("groupPolicy = %q, want %q", result.groupPolicy, "open") + } + if result.requireMention == nil || *result.requireMention != true { + t.Errorf("requireMention = %v, want true", result.requireMention) + } + if len(result.allowFrom) != 1 || result.allowFrom[0] != "user1" { + t.Errorf("allowFrom = %v, want [user1]", result.allowFrom) + } + if !result.isEnabled() { + t.Error("isEnabled() = false, want true") + } +} + +func TestResolveTopicConfig_WildcardGroup(t *testing.T) { + cfg := config.TelegramConfig{ + GroupPolicy: "open", + Groups: map[string]*config.TelegramGroupConfig{ + "*": { + GroupPolicy: "allowlist", + RequireMention: boolPtr(false), + AllowFrom: []string{"admin1"}, + }, + }, + } + + result := resolveTopicConfig(cfg, "-100123", 0) + + if result.groupPolicy != "allowlist" { + t.Errorf("groupPolicy = %q, want %q", result.groupPolicy, "allowlist") + } + if result.requireMention == nil || *result.requireMention != false { + t.Errorf("requireMention = %v, want false", result.requireMention) + } + if len(result.allowFrom) != 1 || result.allowFrom[0] != "admin1" { + t.Errorf("allowFrom = %v, want [admin1]", result.allowFrom) + } +} + +func TestResolveTopicConfig_SpecificGroupOverridesWildcard(t *testing.T) { + cfg := config.TelegramConfig{ + GroupPolicy: "open", + Groups: map[string]*config.TelegramGroupConfig{ + "*": { + GroupPolicy: "allowlist", + RequireMention: boolPtr(false), + AllowFrom: []string{"admin1"}, + SystemPrompt: "wildcard prompt", + }, + "-100123": { + GroupPolicy: "disabled", + AllowFrom: []string{"user2"}, + SystemPrompt: "group prompt", + }, + }, + } + + result := resolveTopicConfig(cfg, "-100123", 0) + + if result.groupPolicy != "disabled" { + t.Errorf("groupPolicy = %q, want %q", result.groupPolicy, "disabled") + } + // requireMention not set on specific group → inherits wildcard + if result.requireMention == nil || *result.requireMention != false { + t.Errorf("requireMention = %v, want false (inherited from wildcard)", result.requireMention) + } + if len(result.allowFrom) != 1 || result.allowFrom[0] != "user2" { + t.Errorf("allowFrom = %v, want [user2]", result.allowFrom) + } + if result.systemPrompt != "group prompt" { + t.Errorf("systemPrompt = %q, want %q", result.systemPrompt, "group prompt") + } +} + +func TestResolveTopicConfig_TopicOverridesGroup(t *testing.T) { + cfg := config.TelegramConfig{ + GroupPolicy: "open", + Groups: map[string]*config.TelegramGroupConfig{ + "-100123": { + RequireMention: boolPtr(true), + SystemPrompt: "group prompt", + Skills: []string{"skill_a", "skill_b"}, + Topics: map[string]*config.TelegramTopicConfig{ + "42": { + RequireMention: boolPtr(false), + Skills: []string{"skill_c"}, + SystemPrompt: "topic prompt", + }, + }, + }, + }, + } + + result := resolveTopicConfig(cfg, "-100123", 42) + + if result.requireMention == nil || *result.requireMention != false { + t.Errorf("requireMention = %v, want false (topic override)", result.requireMention) + } + if len(result.skills) != 1 || result.skills[0] != "skill_c" { + t.Errorf("skills = %v, want [skill_c]", result.skills) + } + // systemPrompt: concatenated group + topic + expected := "group prompt\n\ntopic prompt" + if result.systemPrompt != expected { + t.Errorf("systemPrompt = %q, want %q", result.systemPrompt, expected) + } +} + +func TestResolveTopicConfig_TopicSystemPromptConcatenation(t *testing.T) { + cfg := config.TelegramConfig{ + Groups: map[string]*config.TelegramGroupConfig{ + "-100123": { + SystemPrompt: " group prompt ", + Topics: map[string]*config.TelegramTopicConfig{ + "5": { + SystemPrompt: " topic prompt ", + }, + }, + }, + }, + } + + result := resolveTopicConfig(cfg, "-100123", 5) + + expected := "group prompt\n\ntopic prompt" + if result.systemPrompt != expected { + t.Errorf("systemPrompt = %q, want %q", result.systemPrompt, expected) + } +} + +func TestResolveTopicConfig_TopicOnlySystemPrompt(t *testing.T) { + cfg := config.TelegramConfig{ + Groups: map[string]*config.TelegramGroupConfig{ + "-100123": { + Topics: map[string]*config.TelegramTopicConfig{ + "5": { + SystemPrompt: "topic only", + }, + }, + }, + }, + } + + result := resolveTopicConfig(cfg, "-100123", 5) + + if result.systemPrompt != "topic only" { + t.Errorf("systemPrompt = %q, want %q", result.systemPrompt, "topic only") + } +} + +func TestResolveTopicConfig_DisabledTopic(t *testing.T) { + cfg := config.TelegramConfig{ + Groups: map[string]*config.TelegramGroupConfig{ + "-100123": { + Enabled: boolPtr(true), + Topics: map[string]*config.TelegramTopicConfig{ + "42": { + Enabled: boolPtr(false), + }, + }, + }, + }, + } + + // Group is enabled + groupResult := resolveTopicConfig(cfg, "-100123", 0) + if !groupResult.isEnabled() { + t.Error("group isEnabled() = false, want true") + } + + // Topic 42 is disabled + topicResult := resolveTopicConfig(cfg, "-100123", 42) + if topicResult.isEnabled() { + t.Error("topic isEnabled() = true, want false") + } +} + +func TestResolveTopicConfig_UnknownGroupFallsToWildcard(t *testing.T) { + cfg := config.TelegramConfig{ + Groups: map[string]*config.TelegramGroupConfig{ + "*": { + GroupPolicy: "allowlist", + }, + "-100999": { + GroupPolicy: "disabled", + }, + }, + } + + // Unknown group: gets wildcard + result := resolveTopicConfig(cfg, "-100123", 0) + if result.groupPolicy != "allowlist" { + t.Errorf("groupPolicy = %q, want %q (wildcard)", result.groupPolicy, "allowlist") + } + + // Known group: gets specific + result2 := resolveTopicConfig(cfg, "-100999", 0) + if result2.groupPolicy != "disabled" { + t.Errorf("groupPolicy = %q, want %q (specific)", result2.groupPolicy, "disabled") + } +} + +func TestEffectiveRequireMention(t *testing.T) { + // nil requireMention → use default + r := resolvedTopicConfig{} + if r.effectiveRequireMention(true) != true { + t.Error("effectiveRequireMention(true) = false, want true") + } + if r.effectiveRequireMention(false) != false { + t.Error("effectiveRequireMention(false) = true, want false") + } + + // explicit requireMention overrides default + r2 := resolvedTopicConfig{requireMention: boolPtr(false)} + if r2.effectiveRequireMention(true) != false { + t.Error("effectiveRequireMention(true) with override=false should be false") + } +} + +func TestResolveTopicConfig_EmptySkillsOverride(t *testing.T) { + cfg := config.TelegramConfig{ + Groups: map[string]*config.TelegramGroupConfig{ + "-100123": { + Skills: []string{"skill_a"}, + Topics: map[string]*config.TelegramTopicConfig{ + "42": { + Skills: []string{}, // explicitly empty = no skills + }, + }, + }, + }, + } + + result := resolveTopicConfig(cfg, "-100123", 42) + + if result.skills == nil { + t.Error("skills should be non-nil (explicit empty override)") + } + if len(result.skills) != 0 { + t.Errorf("skills = %v, want empty slice", result.skills) + } +} diff --git a/internal/config/config_channels.go b/internal/config/config_channels.go index 225e2124..2f82670e 100644 --- a/internal/config/config_channels.go +++ b/internal/config/config_channels.go @@ -34,6 +34,33 @@ type TelegramConfig struct { // Optional audio-aware routing: when set, voice/audio inbound messages are routed to this // agent instead of the default channel agent. Requires the named agent to exist in the config. VoiceAgentID string `json:"voice_agent_id,omitempty"` // agent ID to route voice inbound to (e.g. "speaking-agent") + + // Per-group (and per-topic) overrides. Key is chat ID string (e.g. "-100123456") or "*" for wildcard. + // TS ref: channels.telegram.groups in src/config/types.telegram.ts. + Groups map[string]*TelegramGroupConfig `json:"groups,omitempty"` +} + +// TelegramGroupConfig defines per-group overrides for a Telegram channel. +// Matching TS TelegramGroupConfig in src/config/types.telegram.ts. +type TelegramGroupConfig struct { + GroupPolicy string `json:"group_policy,omitempty"` // override group policy for this group + RequireMention *bool `json:"require_mention,omitempty"` // override require_mention for this group + AllowFrom FlexibleStringSlice `json:"allow_from,omitempty"` // override allow_from for this group + Enabled *bool `json:"enabled,omitempty"` // disable bot for this group (default: true) + Skills []string `json:"skills,omitempty"` // skill whitelist (nil = all, [] = none) + SystemPrompt string `json:"system_prompt,omitempty"` // extra system prompt for this group + Topics map[string]*TelegramTopicConfig `json:"topics,omitempty"` // per-topic overrides (key: thread ID string) +} + +// TelegramTopicConfig defines per-topic overrides within a Telegram group. +// Matching TS TelegramTopicConfig in src/config/types.telegram.ts. +type TelegramTopicConfig struct { + RequireMention *bool `json:"require_mention,omitempty"` + GroupPolicy string `json:"group_policy,omitempty"` + Skills []string `json:"skills,omitempty"` + Enabled *bool `json:"enabled,omitempty"` + AllowFrom FlexibleStringSlice `json:"allow_from,omitempty"` + SystemPrompt string `json:"system_prompt,omitempty"` } type DiscordConfig struct { diff --git a/internal/sessions/key.go b/internal/sessions/key.go index 806e5eb3..b40922d4 100644 --- a/internal/sessions/key.go +++ b/internal/sessions/key.go @@ -50,6 +50,14 @@ func BuildGroupTopicSessionKey(agentID, channel, chatID string, topicID int) str return fmt.Sprintf("agent:%s:%s:group:%s:topic:%d", agentID, channel, chatID, topicID) } +// BuildDMThreadSessionKey builds the session key for a DM thread (topic in private chat). +// Preserves message_thread_id for session isolation within the same DM. +// +// agent:{agentId}:{channel}:direct:{peerID}:thread:{threadID} +func BuildDMThreadSessionKey(agentID, channel, peerID string, threadID int) string { + return fmt.Sprintf("agent:%s:%s:direct:%s:thread:%d", agentID, channel, peerID, threadID) +} + // BuildSubagentSessionKey builds the session key for a subagent. // // agent:{agentId}:subagent:{label} diff --git a/internal/tools/telegram_forum.go b/internal/tools/telegram_forum.go new file mode 100644 index 00000000..96b642e8 --- /dev/null +++ b/internal/tools/telegram_forum.go @@ -0,0 +1,101 @@ +package tools + +import ( + "context" + "encoding/json" + "fmt" +) + +// ForumTopicCreator can create forum topics in a Telegram supergroup. +// Implemented by telegram.Channel. +type ForumTopicCreator interface { + CreateForumTopic(ctx context.Context, chatID int64, name string, iconColor int, iconEmojiID string) (threadID int, topicName string, err error) +} + +// ForumTopicCreatorProvider returns a ForumTopicCreator (e.g. from the channel manager). +// Returns nil if no Telegram channel is available. +type ForumTopicCreatorProvider func() ForumTopicCreator + +// CreateForumTopicTool lets agents create forum topics in Telegram supergroups. +type CreateForumTopicTool struct { + provider ForumTopicCreatorProvider +} + +// NewCreateForumTopicTool creates a new create_forum_topic tool. +func NewCreateForumTopicTool(provider ForumTopicCreatorProvider) *CreateForumTopicTool { + return &CreateForumTopicTool{provider: provider} +} + +func (t *CreateForumTopicTool) Name() string { return "create_forum_topic" } + +func (t *CreateForumTopicTool) Description() string { + return "Create a new forum topic in a Telegram supergroup. " + + "Returns the topic's message_thread_id which can be used for routing messages to the topic." +} + +func (t *CreateForumTopicTool) Parameters() map[string]interface{} { + return map[string]interface{}{ + "type": "object", + "properties": map[string]interface{}{ + "chat_id": map[string]interface{}{ + "type": "string", + "description": "The Telegram chat ID of the supergroup (e.g. \"-1001234567890\").", + }, + "name": map[string]interface{}{ + "type": "string", + "description": "Topic name (1-128 characters).", + }, + "icon_color": map[string]interface{}{ + "type": "integer", + "description": "Optional icon color as RGB integer. Allowed values: 7322096, 16766590, 13338331, 9367192, 16749490, 16478047.", + }, + "icon_custom_emoji_id": map[string]interface{}{ + "type": "string", + "description": "Optional custom emoji ID for the topic icon (requires Telegram Premium).", + }, + }, + "required": []string{"chat_id", "name"}, + } +} + +func (t *CreateForumTopicTool) Execute(ctx context.Context, args map[string]interface{}) *Result { + creator := t.provider() + if creator == nil { + return &Result{ForLLM: "Error: no Telegram channel available", IsError: true} + } + + chatIDStr, _ := args["chat_id"].(string) + name, _ := args["name"].(string) + + if chatIDStr == "" || name == "" { + return &Result{ForLLM: "Error: chat_id and name are required", IsError: true} + } + if len(name) > 128 { + return &Result{ForLLM: "Error: topic name must be 1-128 characters", IsError: true} + } + + var chatID int64 + if _, err := fmt.Sscanf(chatIDStr, "%d", &chatID); err != nil { + return &Result{ForLLM: fmt.Sprintf("Error: invalid chat_id %q: %v", chatIDStr, err), IsError: true} + } + + iconColor := 0 + if ic, ok := args["icon_color"].(float64); ok { + iconColor = int(ic) + } + iconEmojiID, _ := args["icon_custom_emoji_id"].(string) + + threadID, topicName, err := creator.CreateForumTopic(ctx, chatID, name, iconColor, iconEmojiID) + if err != nil { + return &Result{ForLLM: fmt.Sprintf("Error creating forum topic: %v", err), IsError: true} + } + + result := map[string]interface{}{ + "message_thread_id": threadID, + "name": topicName, + "chat_id": chatIDStr, + } + + jsonBytes, _ := json.Marshal(result) + return &Result{ForLLM: string(jsonBytes)} +} diff --git a/ui/web/src/pages/channels/channel-instance-form-dialog.tsx b/ui/web/src/pages/channels/channel-instance-form-dialog.tsx index 20b340c6..79f4ec38 100644 --- a/ui/web/src/pages/channels/channel-instance-form-dialog.tsx +++ b/ui/web/src/pages/channels/channel-instance-form-dialog.tsx @@ -22,6 +22,7 @@ import type { AgentData } from "@/types/agent"; import { slugify, isValidSlug } from "@/lib/slug"; import { credentialsSchema, configSchema } from "./channel-schemas"; import { ChannelFields } from "./channel-fields"; +import { TelegramGroupOverrides } from "./telegram-group-overrides"; import { CHANNEL_TYPES } from "@/constants/channels"; interface ChannelInstanceFormDialogProps { @@ -235,6 +236,19 @@ export function ChannelInstanceFormDialog({ )} + {/* Telegram group/topic overrides */} + {channelType === "telegram" && ( + >) ?? {}} + onChange={(groups) => { + setConfigValues((prev) => ({ + ...prev, + groups: Object.keys(groups).length > 0 ? groups : undefined, + })); + }} + /> + )} +
diff --git a/ui/web/src/pages/channels/channel-schemas.ts b/ui/web/src/pages/channels/channel-schemas.ts index ff61a142..72027126 100644 --- a/ui/web/src/pages/channels/channel-schemas.ts +++ b/ui/web/src/pages/channels/channel-schemas.ts @@ -21,7 +21,7 @@ const dmPolicyOptions = [ { value: "disabled", label: "Disabled" }, ]; -const groupPolicyOptions = [ +export const groupPolicyOptions = [ { value: "open", label: "Open (accept all)" }, { value: "pairing", label: "Pairing (require approval)" }, { value: "allowlist", label: "Allowlist only" }, diff --git a/ui/web/src/pages/channels/telegram-group-fields.tsx b/ui/web/src/pages/channels/telegram-group-fields.tsx new file mode 100644 index 00000000..7ed8ac7c --- /dev/null +++ b/ui/web/src/pages/channels/telegram-group-fields.tsx @@ -0,0 +1,156 @@ +import { Label } from "@/components/ui/label"; +import { Textarea } from "@/components/ui/textarea"; +import { + Select, + SelectContent, + SelectItem, + SelectTrigger, + SelectValue, +} from "@/components/ui/select"; +import { groupPolicyOptions } from "./channel-schemas"; + +export interface TelegramGroupConfigValues { + group_policy?: string; + require_mention?: boolean; + enabled?: boolean; + allow_from?: string[]; + skills?: string[]; + system_prompt?: string; +} + +const INHERIT = "__inherit__"; + +const triStateOptions = [ + { value: INHERIT, label: "Inherit (channel default)" }, + { value: "true", label: "Yes" }, + { value: "false", label: "No" }, +]; + +const groupPolicyWithInherit = [ + { value: INHERIT, label: "Inherit (channel default)" }, + ...groupPolicyOptions, +]; + +function triStateValue(val: boolean | undefined): string { + if (val === undefined || val === null) return INHERIT; + return val ? "true" : "false"; +} + +function parseTriState(val: string): boolean | undefined { + if (val === INHERIT || val === "") return undefined; + return val === "true"; +} + +interface Props { + config: TelegramGroupConfigValues; + onChange: (config: TelegramGroupConfigValues) => void; + idPrefix: string; +} + +export function TelegramGroupFields({ config, onChange, idPrefix }: Props) { + const update = (patch: Partial) => { + onChange({ ...config, ...patch }); + }; + + return ( +
+
+ + +
+ +
+ + +
+ +
+ + +
+ +
+ +