feat(telegram): /subagents commands + functional options refactor (#600)

- New /subagents and /subagent <id> 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
This commit is contained in:
viettranx committed 2026-03-31 11:45:25 +07:00
1 parent 2c1ef25392
commit 63878b16ca
7 files changed
+302 -33

No files matched your search

+1 -1
View File
@@ -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 {
+52 -26
View File
@@ -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.
+10
View File
@@ -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 <id> — View task detail\n" +
"/subagents — List subagent tasks\n" +
"/subagent <id> — 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
@@ -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)"},
@@ -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 <id> — 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 <task_id>")
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()
}
@@ -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
}
+12 -6
View File
@@ -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
}