From 63878b16ca25da0b97ed1e2c9400e3f6ddf9ba0a Mon Sep 17 00:00:00 2001 From: viettranx Date: Tue, 31 Mar 2026 11:45:25 +0700 Subject: [PATCH] feat(telegram): /subagents commands + functional options refactor (#600) - New /subagents and /subagent commands for viewing subagent tasks from the persistent DB table - Inline keyboard with sa: callback prefix for detail view - Refactor telegram.New() to functional options pattern (WithAgentStore, WithTeamStore, WithSubagentTaskStore, WithPendingMessageStore) - Wire SubagentTaskStore via WithSubagentTaskStore option --- cmd/gateway_channels_setup.go | 2 +- internal/channels/telegram/channel.go | 78 ++++--- internal/channels/telegram/commands.go | 10 + .../channels/telegram/commands_pairing.go | 2 + .../channels/telegram/commands_subagents.go | 220 ++++++++++++++++++ internal/channels/telegram/commands_tasks.go | 5 + internal/channels/telegram/factory.go | 18 +- 7 files changed, 302 insertions(+), 33 deletions(-) create mode 100644 internal/channels/telegram/commands_subagents.go diff --git a/cmd/gateway_channels_setup.go b/cmd/gateway_channels_setup.go index b3596f8e..c60b34b0 100644 --- a/cmd/gateway_channels_setup.go +++ b/cmd/gateway_channels_setup.go @@ -29,7 +29,7 @@ import ( // registerConfigChannels registers config-based channels as fallback when no DB instances are loaded. func registerConfigChannels(cfg *config.Config, channelMgr *channels.Manager, msgBus *bus.MessageBus, pgStores *store.Stores, instanceLoader *channels.InstanceLoader) { if cfg.Channels.Telegram.Enabled && cfg.Channels.Telegram.Token != "" && instanceLoader == nil { - tg, err := telegram.New(cfg.Channels.Telegram, msgBus, pgStores.Pairing, nil, nil, nil, nil) + tg, err := telegram.New(cfg.Channels.Telegram, msgBus, pgStores.Pairing) if err != nil { slog.Error("failed to initialize telegram channel", "error", err) } else { diff --git a/internal/channels/telegram/channel.go b/internal/channels/telegram/channel.go index d6fa135e..0296535e 100644 --- a/internal/channels/telegram/channel.go +++ b/internal/channels/telegram/channel.go @@ -29,9 +29,10 @@ type Channel struct { transport *http.Transport ipv4Once sync.Once // guards enableIPv4Only to prevent data race pairingService store.PairingStore - agentStore store.AgentStore // for agent key lookup (nil if not configured) - configPermStore store.ConfigPermissionStore // for group file writer management (nil if not configured) - teamStore store.TeamStore // for /tasks, /task_detail commands (nil if not configured) + agentStore store.AgentStore // for agent key lookup (nil if not configured) + configPermStore store.ConfigPermissionStore // for group file writer management (nil if not configured) + teamStore store.TeamStore // for /tasks, /task_detail commands (nil if not configured) + subagentTaskStore store.SubagentTaskStore // for /subagents, /subagent commands (nil if not configured) placeholders sync.Map // localKey string → messageID int stopThinking sync.Map // localKey string → *thinkingCancel typingCtrls sync.Map // localKey string → *typing.Controller @@ -60,16 +61,40 @@ func (c *thinkingCancel) Cancel() { } } +// Option configures optional dependencies for the Telegram channel. +type Option func(*Channel) + +// WithAgentStore sets the agent store for agent key resolution. +func WithAgentStore(s store.AgentStore) Option { return func(c *Channel) { c.agentStore = s } } + +// WithConfigPermStore sets the config permission store for group file writer management. +func WithConfigPermStore(s store.ConfigPermissionStore) Option { + return func(c *Channel) { c.configPermStore = s } +} + +// WithTeamStore sets the team store for /tasks, /task_detail commands. +func WithTeamStore(s store.TeamStore) Option { return func(c *Channel) { c.teamStore = s } } + +// WithSubagentTaskStore sets the subagent task store for /subagents, /subagent commands. +func WithSubagentTaskStore(s store.SubagentTaskStore) Option { + return func(c *Channel) { c.subagentTaskStore = s } +} + +// WithPendingMessageStore sets the pending message store for group history buffering. +func WithPendingMessageStore(s store.PendingMessageStore) Option { + return func(c *Channel) { + c.groupHistory = channels.MakeHistory(channels.TypeTelegram, s, c.TenantID()) + } +} + // New creates a new Telegram channel from config. // pairingSvc is optional (nil = fall back to allowlist only). -// agentStore is optional (nil = group file writer commands disabled). -// configPermStore is optional (nil = group file writer commands disabled). -// teamStore is optional (nil = /tasks, /task_detail commands disabled). -func New(cfg config.TelegramConfig, msgBus *bus.MessageBus, pairingSvc store.PairingStore, agentStore store.AgentStore, configPermStore store.ConfigPermissionStore, teamStore store.TeamStore, pendingStore store.PendingMessageStore) (*Channel, error) { - var opts []telego.BotOption +// Optional stores are set via Option functions. +func New(cfg config.TelegramConfig, msgBus *bus.MessageBus, pairingSvc store.PairingStore, chanOpts ...Option) (*Channel, error) { + var botOpts []telego.BotOption if cfg.APIServer != "" { - opts = append(opts, telego.WithAPIServer(cfg.APIServer)) + botOpts = append(botOpts, telego.WithAPIServer(cfg.APIServer)) } // Isolate transport per account: prevents cross-bot connection pool contention @@ -95,9 +120,9 @@ func New(cfg config.TelegramConfig, msgBus *bus.MessageBus, pairingSvc store.Pai slog.Info("telegram: forced IPv4 for account via config") } - opts = append(opts, telego.WithHTTPClient(httpClient)) + botOpts = append(botOpts, telego.WithHTTPClient(httpClient)) - bot, err := telego.NewBot(cfg.Token, opts...) + bot, err := telego.NewBot(cfg.Token, botOpts...) if err != nil { return nil, fmt.Errorf("create telegram bot: %w", err) } @@ -124,21 +149,22 @@ func New(cfg config.TelegramConfig, msgBus *bus.MessageBus, pairingSvc store.Pai mentionMode = "strict" } - return &Channel{ - BaseChannel: base, - bot: bot, - config: cfg, - httpClient: httpClient, - transport: transport, - pairingService: pairingSvc, - agentStore: agentStore, - configPermStore: configPermStore, - teamStore: teamStore, - groupHistory: channels.MakeHistory(channels.TypeTelegram, pendingStore, base.TenantID()), - historyLimit: historyLimit, - requireMention: requireMention, - mentionMode: mentionMode, - }, nil + ch := &Channel{ + BaseChannel: base, + bot: bot, + config: cfg, + httpClient: httpClient, + transport: transport, + pairingService: pairingSvc, + groupHistory: channels.MakeHistory(channels.TypeTelegram, nil, base.TenantID()), + historyLimit: historyLimit, + requireMention: requireMention, + mentionMode: mentionMode, + } + for _, o := range chanOpts { + o(ch) + } + return ch, nil } // Start begins long polling for Telegram updates. diff --git a/internal/channels/telegram/commands.go b/internal/channels/telegram/commands.go index 877b3875..d87b4bcf 100644 --- a/internal/channels/telegram/commands.go +++ b/internal/channels/telegram/commands.go @@ -87,6 +87,8 @@ func (c *Channel) handleBotCommand(ctx context.Context, message *telego.Message, "/status — Show bot status\n" + "/tasks — List team tasks\n" + "/task_detail — View task detail\n" + + "/subagents — List subagent tasks\n" + + "/subagent — View subagent task detail\n" + "/writers — List file writers for this group\n" + "/addwriter — Add a file writer (reply to their message)\n" + "/removewriter — Remove a file writer (reply to their message)\n" + @@ -205,6 +207,14 @@ func (c *Channel) handleBotCommand(ctx context.Context, message *telego.Message, c.handleTaskDetail(ctx, chatID, text, isGroup, setThread) return true + case "/subagents": + c.handleSubagentsList(ctx, chatID, isGroup, setThread) + return true + + case "/subagent": + c.handleSubagentDetail(ctx, chatID, text, isGroup, setThread) + return true + case "/addwriter": c.handleWriterCommand(ctx, message, chatID, chatIDStr, senderID, isGroup, setThread, "add") return true diff --git a/internal/channels/telegram/commands_pairing.go b/internal/channels/telegram/commands_pairing.go index 9a003495..ecc92074 100644 --- a/internal/channels/telegram/commands_pairing.go +++ b/internal/channels/telegram/commands_pairing.go @@ -154,6 +154,8 @@ func DefaultMenuCommands() []telego.BotCommand { {Command: "status", Description: "Show bot status"}, {Command: "tasks", Description: "List team tasks"}, {Command: "task_detail", Description: "View task detail by ID"}, + {Command: "subagents", Description: "List subagent tasks"}, + {Command: "subagent", Description: "View subagent task detail by ID"}, {Command: "writers", Description: "List file writers for this group"}, {Command: "addwriter", Description: "Add a file writer (reply to their message)"}, {Command: "removewriter", Description: "Remove a file writer (reply to their message)"}, diff --git a/internal/channels/telegram/commands_subagents.go b/internal/channels/telegram/commands_subagents.go new file mode 100644 index 00000000..6bd885cc --- /dev/null +++ b/internal/channels/telegram/commands_subagents.go @@ -0,0 +1,220 @@ +package telegram + +import ( + "context" + "fmt" + "log/slog" + "strings" + + "github.com/google/uuid" + "github.com/mymmrac/telego" + tu "github.com/mymmrac/telego/telegoutil" + + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +const maxSubagentsInList = 30 + +// subagentStatusIcon returns an icon for each subagent task status. +func subagentStatusIcon(status string) string { + switch status { + case "completed": + return "✅" + case "failed": + return "❌" + case "cancelled": + return "⏹" + default: // running + return "🔄" + } +} + +// formatTokenCount formats token counts as "1.2k" for readability. +func formatTokenCount(n int64) string { + if n >= 1000 { + return fmt.Sprintf("%.1fk", float64(n)/1000) + } + return fmt.Sprintf("%d", n) +} + +// handleSubagentsList handles /subagents — lists subagent tasks from DB. +func (c *Channel) handleSubagentsList(ctx context.Context, chatID int64, isGroup bool, setThread func(*telego.SendMessageParams)) { + chatIDObj := tu.ID(chatID) + + send := func(text string) { + msg := tu.Message(chatIDObj, text) + setThread(msg) + c.bot.SendMessage(ctx, msg) + } + + if c.subagentTaskStore == nil { + send("Subagent task tracking is not available.") + return + } + + agentKey := c.AgentID() + if agentKey == "" { + send("Subagent tasks are not available (no agent configured).") + return + } + + tasks, err := c.subagentTaskStore.ListByParent(ctx, agentKey, "") + if err != nil { + slog.Warn("subagents command: ListByParent failed", "error", err) + send("Failed to list subagent tasks. Please try again.") + return + } + + if len(tasks) == 0 { + send("No subagent tasks found.") + return + } + + total := len(tasks) + if total > maxSubagentsInList { + tasks = tasks[:maxSubagentsInList] + } + + var sb strings.Builder + if total > maxSubagentsInList { + sb.WriteString(fmt.Sprintf("Subagent tasks (showing %d of %d):\n\n", maxSubagentsInList, total)) + } else { + sb.WriteString(fmt.Sprintf("Subagent tasks (%d):\n\n", total)) + } + + for i, t := range tasks { + model := "" + if t.Model != nil && *t.Model != "" { + model = *t.Model + } + tokens := fmt.Sprintf("%s/%s tokens", formatTokenCount(t.InputTokens), formatTokenCount(t.OutputTokens)) + if model != "" { + sb.WriteString(fmt.Sprintf("%d. %s %s (%s, %s)\n", i+1, subagentStatusIcon(t.Status), truncateStr(t.Subject, 40), model, tokens)) + } else { + sb.WriteString(fmt.Sprintf("%d. %s %s (%s)\n", i+1, subagentStatusIcon(t.Status), truncateStr(t.Subject, 40), tokens)) + } + } + sb.WriteString("\nTap a button below to view details.") + + var rows [][]telego.InlineKeyboardButton + for i, t := range tasks { + label := fmt.Sprintf("%d. %s %s", i+1, subagentStatusIcon(t.Status), truncateStr(t.Subject, 35)) + rows = append(rows, []telego.InlineKeyboardButton{ + {Text: label, CallbackData: "sa:" + t.ID.String()}, + }) + } + + msg := tu.Message(chatIDObj, sb.String()) + setThread(msg) + if len(rows) > 0 { + msg.ReplyMarkup = &telego.InlineKeyboardMarkup{InlineKeyboard: rows} + } + c.bot.SendMessage(ctx, msg) +} + +// handleSubagentDetail handles /subagent — shows detail for a subagent task. +func (c *Channel) handleSubagentDetail(ctx context.Context, chatID int64, text string, isGroup bool, setThread func(*telego.SendMessageParams)) { + chatIDObj := tu.ID(chatID) + + send := func(t string) { + for _, chunk := range chunkPlainText(t, telegramMaxMessageLen) { + msg := tu.Message(chatIDObj, chunk) + setThread(msg) + c.bot.SendMessage(ctx, msg) + } + } + + parts := strings.SplitN(text, " ", 2) + if len(parts) < 2 || strings.TrimSpace(parts[1]) == "" { + send("Usage: /subagent ") + return + } + idArg := strings.TrimSpace(parts[1]) + + if c.subagentTaskStore == nil { + send("Subagent task tracking is not available.") + return + } + + taskID, err := uuid.Parse(idArg) + if err != nil { + send(fmt.Sprintf("Invalid task ID %q. Use /subagents to list tasks.", idArg)) + return + } + + task, err := c.subagentTaskStore.Get(ctx, taskID) + if err != nil { + slog.Warn("subagent command: Get failed", "id", idArg, "error", err) + send("Failed to load subagent task. Please try again.") + return + } + if task == nil { + send(fmt.Sprintf("Task %q not found. Use /subagents to see available tasks.", idArg[:8])) + return + } + + send(formatSubagentDetail(task)) +} + +// handleSubagentCallback handles "sa:" callback prefix from inline keyboard buttons. +func (c *Channel) handleSubagentCallback(ctx context.Context, query *telego.CallbackQuery) { + taskIDStr := strings.TrimPrefix(query.Data, "sa:") + + chat := query.Message.GetChat() + chatIDObj := tu.ID(chat.ID) + + send := func(text string) { + for _, chunk := range chunkPlainText(text, telegramMaxMessageLen) { + msg := tu.Message(chatIDObj, chunk) + c.bot.SendMessage(ctx, msg) + } + } + + if c.subagentTaskStore == nil { + send("Subagent task tracking is not available.") + return + } + + taskID, err := uuid.Parse(taskIDStr) + if err != nil { + send("Invalid task ID.") + return + } + + task, err := c.subagentTaskStore.Get(ctx, taskID) + if err != nil { + slog.Warn("subagent callback: Get failed", "id", taskIDStr, "error", err) + send("Failed to load subagent task.") + return + } + if task == nil { + send(fmt.Sprintf("Task %s not found.", taskIDStr[:8])) + return + } + + send(formatSubagentDetail(task)) +} + +// formatSubagentDetail formats a single subagent task for display. +func formatSubagentDetail(t *store.SubagentTaskData) string { + var sb strings.Builder + sb.WriteString(fmt.Sprintf("Subagent: %s\n", t.Subject)) + sb.WriteString(fmt.Sprintf("ID: %s\n", t.ID.String())) + sb.WriteString(fmt.Sprintf("Status: %s %s\n", subagentStatusIcon(t.Status), t.Status)) + if t.Model != nil && *t.Model != "" { + sb.WriteString(fmt.Sprintf("Model: %s\n", *t.Model)) + } + sb.WriteString(fmt.Sprintf("Depth: %d\n", t.Depth)) + sb.WriteString(fmt.Sprintf("Iterations: %d\n", t.Iterations)) + sb.WriteString(fmt.Sprintf("Tokens: %s in / %s out\n", formatTokenCount(t.InputTokens), formatTokenCount(t.OutputTokens))) + if !t.CreatedAt.IsZero() { + sb.WriteString(fmt.Sprintf("Created: %s\n", t.CreatedAt.Format("2006-01-02 15:04"))) + } + if t.Description != "" { + sb.WriteString(fmt.Sprintf("\nPrompt:\n%s\n", truncateStr(t.Description, 500))) + } + if t.Result != nil && *t.Result != "" { + sb.WriteString(fmt.Sprintf("\nResult:\n%s\n", truncateStr(*t.Result, 1000))) + } + return sb.String() +} diff --git a/internal/channels/telegram/commands_tasks.go b/internal/channels/telegram/commands_tasks.go index 02c0dfde..fe58bd8c 100644 --- a/internal/channels/telegram/commands_tasks.go +++ b/internal/channels/telegram/commands_tasks.go @@ -203,6 +203,11 @@ func (c *Channel) handleCallbackQuery(ctx context.Context, query *telego.Callbac CallbackQueryID: query.ID, }) + if strings.HasPrefix(query.Data, "sa:") { + c.handleSubagentCallback(ctx, query) + return + } + if !strings.HasPrefix(query.Data, "td:") { return } diff --git a/internal/channels/telegram/factory.go b/internal/channels/telegram/factory.go index 786e2ac0..f627b4d2 100644 --- a/internal/channels/telegram/factory.go +++ b/internal/channels/telegram/factory.go @@ -42,19 +42,25 @@ type telegramInstanceConfig struct { // Factory creates a Telegram channel from DB instance data (no extra stores). func Factory(name string, creds json.RawMessage, cfg json.RawMessage, msgBus *bus.MessageBus, pairingSvc store.PairingStore) (channels.Channel, error) { - return buildChannel(name, creds, cfg, msgBus, pairingSvc, nil, nil, nil, nil) + return buildChannel(name, creds, cfg, msgBus, pairingSvc) } -// FactoryWithStores returns a ChannelFactory that includes agent, configPerm, team, and pending message stores. -func FactoryWithStores(agentStore store.AgentStore, configPermStore store.ConfigPermissionStore, teamStore store.TeamStore, pendingStore store.PendingMessageStore) channels.ChannelFactory { +// FactoryWithStores returns a ChannelFactory that includes optional stores via functional options. +func FactoryWithStores(agentStore store.AgentStore, configPermStore store.ConfigPermissionStore, teamStore store.TeamStore, subagentTaskStore store.SubagentTaskStore, pendingStore store.PendingMessageStore) channels.ChannelFactory { return func(name string, creds json.RawMessage, cfg json.RawMessage, msgBus *bus.MessageBus, pairingSvc store.PairingStore) (channels.Channel, error) { - return buildChannel(name, creds, cfg, msgBus, pairingSvc, agentStore, configPermStore, teamStore, pendingStore) + return buildChannel(name, creds, cfg, msgBus, pairingSvc, + WithAgentStore(agentStore), + WithConfigPermStore(configPermStore), + WithTeamStore(teamStore), + WithSubagentTaskStore(subagentTaskStore), + WithPendingMessageStore(pendingStore), + ) } } func buildChannel(name string, creds json.RawMessage, cfg json.RawMessage, - msgBus *bus.MessageBus, pairingSvc store.PairingStore, agentStore store.AgentStore, configPermStore store.ConfigPermissionStore, teamStore store.TeamStore, pendingStore store.PendingMessageStore) (channels.Channel, error) { + msgBus *bus.MessageBus, pairingSvc store.PairingStore, opts ...Option) (channels.Channel, error) { var c telegramCreds if len(creds) > 0 { @@ -111,7 +117,7 @@ func buildChannel(name string, creds json.RawMessage, cfg json.RawMessage, tgCfg.GroupPolicy = "pairing" } - ch, err := New(tgCfg, msgBus, pairingSvc, agentStore, configPermStore, teamStore, pendingStore) + ch, err := New(tgCfg, msgBus, pairingSvc, opts...) if err != nil { return nil, err }