diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index d809981b4..f9b4b80f4 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -467,47 +467,32 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, agent *AgentInstance, opt } func (al *AgentLoop) targetReasoningChannelID(channelName string) (chatID string) { - channels := al.cfg.Channels - - switch channelName { - case "telegram": - return channels.Telegram.ReasoningChannelID - case "whatsapp": - return channels.WhatsApp.ReasoningChannelID - case "feishu": - return channels.Feishu.ReasoningChannelID - case "discord": - return channels.Discord.ReasoningChannelID - case "maixcam": - return channels.MaixCam.ReasoningChannelID - case "qq": - return channels.QQ.ReasoningChannelID - case "dingtalk": - return channels.DingTalk.ReasoningChannelID - case "slack": - return channels.Slack.ReasoningChannelID - case "line": - return channels.LINE.ReasoningChannelID - case "onebot": - return channels.OneBot.ReasoningChannelID - case "wecom": - return channels.WeCom.ReasoningChannelID - case "wecom_app": - return channels.WeComApp.ReasoningChannelID + if al.channelManager == nil { + return "" } - return + + if ch, ok := al.channelManager.GetChannel(channelName); ok { + return ch.ReasoningChannelID() + } + + return "" } -func (al *AgentLoop) handleReasoning(reasoningContent, channelName, channelID string) { +func (al *AgentLoop) handleReasoning(ctx context.Context, reasoningContent, channelName, channelID string) { if reasoningContent == "" || channelName == "" || channelID == "" { return } - al.bus.PublishOutbound(bus.OutboundMessage{ - Channel: channelName, - ChatID: channelID, - Content: reasoningContent, - }) + select { + case <-ctx.Done(): + return + default: + al.bus.PublishOutbound(bus.OutboundMessage{ + Channel: channelName, + ChatID: channelID, + Content: reasoningContent, + }) + } } // runLLMIteration executes the LLM call loop with tool handling. @@ -634,7 +619,7 @@ func (al *AgentLoop) runLLMIteration( return "", iteration, fmt.Errorf("LLM call failed after retries: %w", err) } - go al.handleReasoning(response.Reasoning, opts.Channel, al.targetReasoningChannelID(opts.Channel)) + go al.handleReasoning(ctx, response.Reasoning, opts.Channel, al.targetReasoningChannelID(opts.Channel)) // Log LLM response details logger.InfoCF("agent", "LLM response", map[string]any{ diff --git a/pkg/agent/loop_test.go b/pkg/agent/loop_test.go index 4c5020b52..96bc24ea0 100644 --- a/pkg/agent/loop_test.go +++ b/pkg/agent/loop_test.go @@ -9,11 +9,22 @@ import ( "time" "github.com/sipeed/picoclaw/pkg/bus" + "github.com/sipeed/picoclaw/pkg/channels" "github.com/sipeed/picoclaw/pkg/config" "github.com/sipeed/picoclaw/pkg/providers" "github.com/sipeed/picoclaw/pkg/tools" ) +type fakeChannel struct{ id string } + +func (f *fakeChannel) Name() string { return "fake" } +func (f *fakeChannel) Start(ctx context.Context) error { return nil } +func (f *fakeChannel) Stop(ctx context.Context) error { return nil } +func (f *fakeChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { return nil } +func (f *fakeChannel) IsRunning() bool { return true } +func (f *fakeChannel) IsAllowed(string) bool { return true } +func (f *fakeChannel) ReasoningChannelID() string { return f.id } + func TestRecordLastChannel(t *testing.T) { // Create temp workspace tmpDir, err := os.MkdirTemp("", "agent-test-*") @@ -648,23 +659,30 @@ func TestTargetReasoningChannelID_AllChannels(t *testing.T) { MaxToolIterations: 10, }, }, - Channels: config.ChannelsConfig{ - WhatsApp: config.WhatsAppConfig{ReasoningChannelID: "rid-whatsapp"}, - Telegram: config.TelegramConfig{ReasoningChannelID: "rid-telegram"}, - Feishu: config.FeishuConfig{ReasoningChannelID: "rid-feishu"}, - Discord: config.DiscordConfig{ReasoningChannelID: "rid-discord"}, - MaixCam: config.MaixCamConfig{ReasoningChannelID: "rid-maixcam"}, - QQ: config.QQConfig{ReasoningChannelID: "rid-qq"}, - DingTalk: config.DingTalkConfig{ReasoningChannelID: "rid-dingtalk"}, - Slack: config.SlackConfig{ReasoningChannelID: "rid-slack"}, - LINE: config.LINEConfig{ReasoningChannelID: "rid-line"}, - OneBot: config.OneBotConfig{ReasoningChannelID: "rid-onebot"}, - WeCom: config.WeComConfig{ReasoningChannelID: "rid-wecom"}, - WeComApp: config.WeComAppConfig{ReasoningChannelID: "rid-wecom-app"}, - }, } al := NewAgentLoop(cfg, bus.NewMessageBus(), &mockProvider{}) + chManager, err := channels.NewManager(&config.Config{}, bus.NewMessageBus()) + if err != nil { + t.Fatalf("Failed to create channel manager: %v", err) + } + for name, id := range map[string]string{ + "whatsapp": "rid-whatsapp", + "telegram": "rid-telegram", + "feishu": "rid-feishu", + "discord": "rid-discord", + "maixcam": "rid-maixcam", + "qq": "rid-qq", + "dingtalk": "rid-dingtalk", + "slack": "rid-slack", + "line": "rid-line", + "onebot": "rid-onebot", + "wecom": "rid-wecom", + "wecom_app": "rid-wecom-app", + } { + chManager.RegisterChannel(name, &fakeChannel{id: id}) + } + al.SetChannelManager(chManager) tests := []struct { channel string wantID string @@ -718,7 +736,7 @@ func TestHandleReasoning(t *testing.T) { t.Run("skips when any required field is empty", func(t *testing.T) { al, msgBus := newLoop(t) - al.handleReasoning("reasoning", "telegram", "") + al.handleReasoning(context.Background(), "reasoning", "telegram", "") ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) defer cancel() @@ -729,7 +747,7 @@ func TestHandleReasoning(t *testing.T) { t.Run("publishes one message for non telegram", func(t *testing.T) { al, msgBus := newLoop(t) - al.handleReasoning("hello reasoning", "slack", "channel-1") + al.handleReasoning(context.Background(), "hello reasoning", "slack", "channel-1") ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond) defer cancel() @@ -745,7 +763,7 @@ func TestHandleReasoning(t *testing.T) { t.Run("publishes one message for telegram", func(t *testing.T) { al, msgBus := newLoop(t) reasoning := "hello telegram reasoning" - al.handleReasoning(reasoning, "telegram", "tg-chat") + al.handleReasoning(context.Background(), reasoning, "telegram", "tg-chat") ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond) defer cancel() @@ -764,4 +782,18 @@ func TestHandleReasoning(t *testing.T) { t.Fatalf("content mismatch: got %q want %q", msg.Content, reasoning) } }) + t.Run("expired ctx", func(t *testing.T) { + al, msgBus := newLoop(t) + reasoning := "hello telegram reasoning" + ctx, cancel := context.WithCancel(context.Background()) + cancel() + al.handleReasoning(ctx, reasoning, "telegram", "tg-chat") + + ctx, cancel = context.WithTimeout(context.Background(), 200*time.Millisecond) + defer cancel() + msg, ok := msgBus.SubscribeOutbound(ctx) + if ok { + t.Fatalf("expected no outbound message, got %+v", msg) + } + }) } diff --git a/pkg/channels/base.go b/pkg/channels/base.go index cd6419ebb..9fe8ec51f 100644 --- a/pkg/channels/base.go +++ b/pkg/channels/base.go @@ -14,23 +14,32 @@ type Channel interface { Send(ctx context.Context, msg bus.OutboundMessage) error IsRunning() bool IsAllowed(senderID string) bool + ReasoningChannelID() string } type BaseChannel struct { - config any - bus *bus.MessageBus - running bool - name string - allowList []string + config any + bus *bus.MessageBus + running bool + name string + allowList []string + reasoningChannelID string } -func NewBaseChannel(name string, config any, bus *bus.MessageBus, allowList []string) *BaseChannel { +func NewBaseChannel( + name string, + config any, + bus *bus.MessageBus, + allowList []string, + reasoningChannelID string, +) *BaseChannel { return &BaseChannel{ - config: config, - bus: bus, - name: name, - allowList: allowList, - running: false, + config: config, + bus: bus, + name: name, + allowList: allowList, + reasoningChannelID: reasoningChannelID, + running: false, } } @@ -38,6 +47,10 @@ func (c *BaseChannel) Name() string { return c.name } +func (c *BaseChannel) ReasoningChannelID() string { + return c.reasoningChannelID +} + func (c *BaseChannel) IsRunning() bool { return c.running } @@ -81,7 +94,11 @@ func (c *BaseChannel) IsAllowed(senderID string) bool { return false } -func (c *BaseChannel) HandleMessage(senderID, chatID, content string, media []string, metadata map[string]string) { +func (c *BaseChannel) HandleMessage( + senderID, chatID, content string, + media []string, + metadata map[string]string, +) { if !c.IsAllowed(senderID) { return } diff --git a/pkg/channels/base_test.go b/pkg/channels/base_test.go index 78c6d1d66..5ee660fc7 100644 --- a/pkg/channels/base_test.go +++ b/pkg/channels/base_test.go @@ -43,7 +43,7 @@ func TestBaseChannelIsAllowed(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - ch := NewBaseChannel("test", nil, nil, tt.allowList) + ch := NewBaseChannel("test", nil, nil, tt.allowList, "") if got := ch.IsAllowed(tt.senderID); got != tt.want { t.Fatalf("IsAllowed(%q) = %v, want %v", tt.senderID, got, tt.want) } diff --git a/pkg/channels/dingtalk.go b/pkg/channels/dingtalk.go index 662fba3b7..c622a406d 100644 --- a/pkg/channels/dingtalk.go +++ b/pkg/channels/dingtalk.go @@ -37,7 +37,7 @@ func NewDingTalkChannel(cfg config.DingTalkConfig, messageBus *bus.MessageBus) ( return nil, fmt.Errorf("dingtalk client_id and client_secret are required") } - base := NewBaseChannel("dingtalk", cfg, messageBus, cfg.AllowFrom) + base := NewBaseChannel("dingtalk", cfg, messageBus, cfg.AllowFrom, cfg.ReasoningChannelID) return &DingTalkChannel{ BaseChannel: base, diff --git a/pkg/channels/discord.go b/pkg/channels/discord.go index 20f3b267c..18e58db84 100644 --- a/pkg/channels/discord.go +++ b/pkg/channels/discord.go @@ -39,7 +39,7 @@ func NewDiscordChannel(cfg config.DiscordConfig, bus *bus.MessageBus) (*DiscordC return nil, fmt.Errorf("failed to create discord session: %w", err) } - base := NewBaseChannel("discord", cfg, bus, cfg.AllowFrom) + base := NewBaseChannel("discord", cfg, bus, cfg.AllowFrom, cfg.ReasoningChannelID) return &DiscordChannel{ BaseChannel: base, diff --git a/pkg/channels/feishu_64.go b/pkg/channels/feishu_64.go index 42e74980f..953ccf9b6 100644 --- a/pkg/channels/feishu_64.go +++ b/pkg/channels/feishu_64.go @@ -31,7 +31,7 @@ type FeishuChannel struct { } func NewFeishuChannel(cfg config.FeishuConfig, bus *bus.MessageBus) (*FeishuChannel, error) { - base := NewBaseChannel("feishu", cfg, bus, cfg.AllowFrom) + base := NewBaseChannel("feishu", cfg, bus, cfg.AllowFrom, cfg.ReasoningChannelID) return &FeishuChannel{ BaseChannel: base, diff --git a/pkg/channels/line.go b/pkg/channels/line.go index 44134996f..3b03b267a 100644 --- a/pkg/channels/line.go +++ b/pkg/channels/line.go @@ -59,7 +59,7 @@ func NewLINEChannel(cfg config.LINEConfig, messageBus *bus.MessageBus) (*LINECha return nil, fmt.Errorf("line channel_secret and channel_access_token are required") } - base := NewBaseChannel("line", cfg, messageBus, cfg.AllowFrom) + base := NewBaseChannel("line", cfg, messageBus, cfg.AllowFrom, cfg.ReasoningChannelID) return &LINEChannel{ BaseChannel: base, diff --git a/pkg/channels/maixcam.go b/pkg/channels/maixcam.go index 34ce62b20..79a1863aa 100644 --- a/pkg/channels/maixcam.go +++ b/pkg/channels/maixcam.go @@ -28,7 +28,7 @@ type MaixCamMessage struct { } func NewMaixCamChannel(cfg config.MaixCamConfig, bus *bus.MessageBus) (*MaixCamChannel, error) { - base := NewBaseChannel("maixcam", cfg, bus, cfg.AllowFrom) + base := NewBaseChannel("maixcam", cfg, bus, cfg.AllowFrom, cfg.ReasoningChannelID) return &MaixCamChannel{ BaseChannel: base, diff --git a/pkg/channels/onebot.go b/pkg/channels/onebot.go index cee8ad9d3..0e970fd3f 100644 --- a/pkg/channels/onebot.go +++ b/pkg/channels/onebot.go @@ -98,7 +98,7 @@ type oneBotMessageSegment struct { } func NewOneBotChannel(cfg config.OneBotConfig, messageBus *bus.MessageBus) (*OneBotChannel, error) { - base := NewBaseChannel("onebot", cfg, messageBus, cfg.AllowFrom) + base := NewBaseChannel("onebot", cfg, messageBus, cfg.AllowFrom, cfg.ReasoningChannelID) const dedupSize = 1024 return &OneBotChannel{ diff --git a/pkg/channels/qq.go b/pkg/channels/qq.go index e66cac533..4d507dc60 100644 --- a/pkg/channels/qq.go +++ b/pkg/channels/qq.go @@ -31,7 +31,7 @@ type QQChannel struct { } func NewQQChannel(cfg config.QQConfig, messageBus *bus.MessageBus) (*QQChannel, error) { - base := NewBaseChannel("qq", cfg, messageBus, cfg.AllowFrom) + base := NewBaseChannel("qq", cfg, messageBus, cfg.AllowFrom, cfg.ReasoningChannelID) return &QQChannel{ BaseChannel: base, diff --git a/pkg/channels/slack.go b/pkg/channels/slack.go index f7359cd6d..baae1b994 100644 --- a/pkg/channels/slack.go +++ b/pkg/channels/slack.go @@ -49,7 +49,7 @@ func NewSlackChannel(cfg config.SlackConfig, messageBus *bus.MessageBus) (*Slack socketClient := socketmode.New(api) - base := NewBaseChannel("slack", cfg, messageBus, cfg.AllowFrom) + base := NewBaseChannel("slack", cfg, messageBus, cfg.AllowFrom, cfg.ReasoningChannelID) return &SlackChannel{ BaseChannel: base, diff --git a/pkg/channels/telegram.go b/pkg/channels/telegram.go index a0a1c8d0a..27d4f3337 100644 --- a/pkg/channels/telegram.go +++ b/pkg/channels/telegram.go @@ -72,7 +72,7 @@ func NewTelegramChannel(cfg *config.Config, bus *bus.MessageBus) (*TelegramChann return nil, fmt.Errorf("failed to create telegram bot: %w", err) } - base := NewBaseChannel("telegram", telegramCfg, bus, telegramCfg.AllowFrom) + base := NewBaseChannel("telegram", telegramCfg, bus, telegramCfg.AllowFrom, telegramCfg.ReasoningChannelID) return &TelegramChannel{ BaseChannel: base, diff --git a/pkg/channels/wecom.go b/pkg/channels/wecom.go index f8daf89de..ea25b5a6f 100644 --- a/pkg/channels/wecom.go +++ b/pkg/channels/wecom.go @@ -96,7 +96,7 @@ func NewWeComBotChannel(cfg config.WeComConfig, messageBus *bus.MessageBus) (*We return nil, fmt.Errorf("wecom token and webhook_url are required") } - base := NewBaseChannel("wecom", cfg, messageBus, cfg.AllowFrom) + base := NewBaseChannel("wecom", cfg, messageBus, cfg.AllowFrom, cfg.ReasoningChannelID) return &WeComBotChannel{ BaseChannel: base, diff --git a/pkg/channels/wecom_app.go b/pkg/channels/wecom_app.go index 715c48707..dde6c174a 100644 --- a/pkg/channels/wecom_app.go +++ b/pkg/channels/wecom_app.go @@ -123,7 +123,7 @@ func NewWeComAppChannel(cfg config.WeComAppConfig, messageBus *bus.MessageBus) ( return nil, fmt.Errorf("wecom_app corp_id, corp_secret and agent_id are required") } - base := NewBaseChannel("wecom_app", cfg, messageBus, cfg.AllowFrom) + base := NewBaseChannel("wecom_app", cfg, messageBus, cfg.AllowFrom, cfg.ReasoningChannelID) return &WeComAppChannel{ BaseChannel: base, diff --git a/pkg/channels/whatsapp.go b/pkg/channels/whatsapp.go index 958d850bb..68ff15a1b 100644 --- a/pkg/channels/whatsapp.go +++ b/pkg/channels/whatsapp.go @@ -25,7 +25,7 @@ type WhatsAppChannel struct { } func NewWhatsAppChannel(cfg config.WhatsAppConfig, bus *bus.MessageBus) (*WhatsAppChannel, error) { - base := NewBaseChannel("whatsapp", cfg, bus, cfg.AllowFrom) + base := NewBaseChannel("whatsapp", cfg, bus, cfg.AllowFrom, cfg.ReasoningChannelID) return &WhatsAppChannel{ BaseChannel: base, diff --git a/pkg/providers/openai_compat/provider.go b/pkg/providers/openai_compat/provider.go index 58076c42f..247f34432 100644 --- a/pkg/providers/openai_compat/provider.go +++ b/pkg/providers/openai_compat/provider.go @@ -25,6 +25,7 @@ type ( ToolFunctionDefinition = protocoltypes.ToolFunctionDefinition ExtraContent = protocoltypes.ExtraContent GoogleExtra = protocoltypes.GoogleExtra + ReasoningDetail = protocoltypes.ReasoningDetail ) type Provider struct { @@ -148,15 +149,10 @@ func parseResponse(body []byte) (*LLMResponse, error) { var apiResponse struct { Choices []struct { Message struct { - Reasoning string `json:"reasoning"` - ReasoningDetails []struct { - Format string `json:"format"` - Index int `json:"index"` - Type string `json:"type"` - Text string `json:"text"` - } `json:"reasoning_details"` - Content string `json:"content"` - ToolCalls []struct { + Reasoning string `json:"reasoning"` + ReasoningDetails []ReasoningDetail `json:"reasoning_details"` + Content string `json:"content"` + ToolCalls []struct { ID string `json:"id"` Type string `json:"type"` Function *struct { diff --git a/pkg/providers/protocoltypes/types.go b/pkg/providers/protocoltypes/types.go index 8d8a08179..a2b4c75c0 100644 --- a/pkg/providers/protocoltypes/types.go +++ b/pkg/providers/protocoltypes/types.go @@ -25,17 +25,19 @@ type FunctionCall struct { } type LLMResponse struct { - Content string `json:"content"` - ToolCalls []ToolCall `json:"tool_calls,omitempty"` - FinishReason string `json:"finish_reason"` - Usage *UsageInfo `json:"usage,omitempty"` - Reasoning string `json:"reasoning"` - ReasoningDetails []struct { - Format string `json:"format"` - Index int `json:"index"` - Type string `json:"type"` - Text string `json:"text"` - } `json:"reasoning_details"` + Content string `json:"content"` + ToolCalls []ToolCall `json:"tool_calls,omitempty"` + FinishReason string `json:"finish_reason"` + Usage *UsageInfo `json:"usage,omitempty"` + Reasoning string `json:"reasoning"` + ReasoningDetails []ReasoningDetail `json:"reasoning_details"` +} + +type ReasoningDetail struct { + Format string `json:"format"` + Index int `json:"index"` + Type string `json:"type"` + Text string `json:"text"` } type UsageInfo struct { PromptTokens int `json:"prompt_tokens"`