diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index cd4276155..c260c13f8 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -19,7 +19,6 @@ import ( "unicode/utf8" "github.com/sipeed/picoclaw/pkg/bus" - "github.com/sipeed/picoclaw/pkg/channels" "github.com/sipeed/picoclaw/pkg/config" "github.com/sipeed/picoclaw/pkg/constants" "github.com/sipeed/picoclaw/pkg/logger" @@ -43,7 +42,6 @@ type AgentLoop struct { tools *tools.ToolRegistry running atomic.Bool summarizing sync.Map // Tracks which sessions are currently being summarized - channelManager *channels.Manager } // processOptions configures how a message is processed @@ -101,6 +99,17 @@ func createToolRegistry(workspace string, restrict bool, cfg *config.Config, msg }) registry.Register(messageTool) + // Pushover tool - send push notifications to phone + pushoverTool := tools.NewPushoverTool() + pushoverTool.SetPushoverCallback(func(message string) error { + msgBus.PublishOutbound(bus.OutboundMessage{ + Channel: "pushover", + Content: message, + }) + return nil + }) + registry.Register(pushoverTool) + return registry } @@ -201,10 +210,6 @@ func (al *AgentLoop) RegisterTool(tool tools.Tool) { al.tools.Register(tool) } -func (al *AgentLoop) SetChannelManager(cm *channels.Manager) { - al.channelManager = cm -} - // RecordLastChannel records the last active channel for this workspace. // This uses the atomic state save mechanism to prevent data loss on crash. func (al *AgentLoop) RecordLastChannel(channel string) error { @@ -269,11 +274,6 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage) return al.processSystemMessage(ctx, msg) } - // Check for commands - if response, handled := al.handleCommand(ctx, msg); handled { - return response, nil - } - // Process as user message return al.runAgentLoop(ctx, processOptions{ SessionKey: msg.SessionKey, @@ -394,7 +394,7 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, opts processOptions) (str // 7. Optional: summarization if opts.EnableSummary { - al.maybeSummarize(opts.SessionKey, opts.Channel, opts.ChatID) + al.maybeSummarize(opts.SessionKey) } // 8. Optional: send response via bus @@ -456,131 +456,11 @@ func (al *AgentLoop) runLLMIteration(ctx context.Context, messages []providers.M "tools_json": formatToolsForLog(providerToolDefs), }) - var response *providers.LLMResponse - var err error - - // Retry loop for context/token errors - maxRetries := 2 - for retry := 0; retry <= maxRetries; retry++ { - response, err = al.provider.Chat(ctx, messages, providerToolDefs, al.model, map[string]interface{}{ - "max_tokens": 8192, - "temperature": 0.7, - }) - - if err == nil { - break // Success - } - - errMsg := strings.ToLower(err.Error()) - // Check for context window errors (provider specific, but usually contain "token" or "invalid") - isContextError := strings.Contains(errMsg, "token") || - strings.Contains(errMsg, "context") || - strings.Contains(errMsg, "invalidparameter") || - strings.Contains(errMsg, "length") - - if isContextError && retry < maxRetries { - logger.WarnCF("agent", "Context window error detected, attempting compression", map[string]interface{}{ - "error": err.Error(), - "retry": retry, - }) - - // Notify user on first retry only - if retry == 0 && !constants.IsInternalChannel(opts.Channel) && opts.SendResponse { - al.bus.PublishOutbound(bus.OutboundMessage{ - Channel: opts.Channel, - ChatID: opts.ChatID, - Content: "⚠️ Context window exceeded. Compressing history and retrying...", - }) - } - - // Force compression - al.forceCompression(opts.SessionKey) - - // Rebuild messages with compressed history - // Note: We need to reload history from session manager because forceCompression changed it - newHistory := al.sessions.GetHistory(opts.SessionKey) - newSummary := al.sessions.GetSummary(opts.SessionKey) - - // Re-create messages for the next attempt - // We keep the current user message (opts.UserMessage) effectively - messages = al.contextBuilder.BuildMessages( - newHistory, - newSummary, - opts.UserMessage, - nil, - opts.Channel, - opts.ChatID, - ) - - // Important: If we are in the middle of a tool loop (iteration > 1), - // rebuilding messages from session history might duplicate the flow or miss context - // if intermediate steps weren't saved correctly. - // However, al.sessions.AddFullMessage is called after every tool execution, - // so GetHistory should reflect the current state including partial tool execution. - // But we need to ensure we don't duplicate the user message which is appended in BuildMessages. - // BuildMessages(history...) takes the stored history and appends the *current* user message. - // If iteration > 1, the "current user message" was already added to history in step 3 of runAgentLoop. - // So if we pass opts.UserMessage again, we might duplicate it? - // Actually, step 3 is: al.sessions.AddMessage(opts.SessionKey, "user", opts.UserMessage) - // So GetHistory ALREADY contains the user message! - - // CORRECTION: - // BuildMessages combines: [System] + [History] + [CurrentMessage] - // But Step 3 added CurrentMessage to History. - // So if we use GetHistory now, it has the user message. - // If we pass opts.UserMessage to BuildMessages, it adds it AGAIN. - - // For retry in the middle of a loop, we should rely on what's in the session. - // BUT checking BuildMessages implementation: - // It appends history... then appends currentMessage. - - // Logic fix for retry: - // If iteration == 1, opts.UserMessage corresponds to the user input. - // If iteration > 1, we are processing tool results. The "messages" passed to Chat - // already accumulated tool outputs. - // Rebuilding from session history is safest because it persists state. - // Start fresh with rebuilt history. - - // Special case: standard BuildMessages appends "currentMessage". - // If we are strictly retrying the *LLM call*, we want the exact same state as before but compressed. - // However, the "messages" argument passed to runLLMIteration is constructed by the caller. - // If we rebuild from Session, we need to know if "currentMessage" should be appended or is already in history. - - // In runAgentLoop: - // 3. sessions.AddMessage(userMsg) - // 4. runLLMIteration(..., UserMessage) - - // So History contains the user message. - // BuildMessages typically appends the user message as a *new* pending message. - // Wait, standard BuildMessages usage in runAgentLoop: - // messages := BuildMessages(history (has old), UserMessage) - // THEN AddMessage(UserMessage). - // So "history" passed to BuildMessages does NOT contain the current UserMessage yet. - - // But here, inside the loop, we have already saved it. - // So GetHistory() includes the current user message. - // If we call BuildMessages(GetHistory(), UserMessage), we get duplicates. - - // Hack/Fix: - // If we are retrying, we rebuild from Session History ONLY. - // We pass empty string as "currentMessage" to BuildMessages - // because the "current message" is already saved in history (step 3). - - messages = al.contextBuilder.BuildMessages( - newHistory, - newSummary, - "", // Empty because history already contains the relevant messages - nil, - opts.Channel, - opts.ChatID, - ) - - continue - } - - // Real error or success, break loop - break - } + // Call LLM + response, err := al.provider.Chat(ctx, messages, providerToolDefs, al.model, map[string]interface{}{ + "max_tokens": 8192, + "temperature": 0.7, + }) if err != nil { logger.ErrorCF("agent", "LLM call failed", @@ -588,7 +468,7 @@ func (al *AgentLoop) runLLMIteration(ctx context.Context, messages []providers.M "iteration": iteration, "error": err.Error(), }) - return "", iteration, fmt.Errorf("LLM call failed after retries: %w", err) + return "", iteration, fmt.Errorf("LLM call failed: %w", err) } // Check if no tool calls - we're done @@ -720,7 +600,7 @@ func (al *AgentLoop) updateToolContexts(channel, chatID string) { } // maybeSummarize triggers summarization if the session history exceeds thresholds. -func (al *AgentLoop) maybeSummarize(sessionKey, channel, chatID string) { +func (al *AgentLoop) maybeSummarize(sessionKey string) { newHistory := al.sessions.GetHistory(sessionKey) tokenEstimate := al.estimateTokens(newHistory) threshold := al.contextWindow * 75 / 100 @@ -729,80 +609,12 @@ func (al *AgentLoop) maybeSummarize(sessionKey, channel, chatID string) { if _, loading := al.summarizing.LoadOrStore(sessionKey, true); !loading { go func() { defer al.summarizing.Delete(sessionKey) - // Notify user about optimization if not an internal channel - if !constants.IsInternalChannel(channel) { - al.bus.PublishOutbound(bus.OutboundMessage{ - Channel: channel, - ChatID: chatID, - Content: "⚠️ Memory threshold reached. Optimizing conversation history...", - }) - } al.summarizeSession(sessionKey) }() } } } -// forceCompression aggressively reduces context when the limit is hit. -// It drops the oldest 50% of messages (keeping system prompt and last user message). -func (al *AgentLoop) forceCompression(sessionKey string) { - history := al.sessions.GetHistory(sessionKey) - if len(history) <= 4 { - return - } - - // Keep system prompt (usually [0]) and the very last message (user's trigger) - // We want to drop the oldest half of the *conversation* - // Assuming [0] is system, [1:] is conversation - conversation := history[1 : len(history)-1] - if len(conversation) == 0 { - return - } - - // Helper to find the mid-point of the conversation - mid := len(conversation) / 2 - - // New history structure: - // 1. System Prompt - // 2. [Summary of dropped part] - synthesized - // 3. Second half of conversation - // 4. Last message - - // Simplified approach for emergency: Drop first half of conversation - // and rely on existing summary if present, or create a placeholder. - - droppedCount := mid - keptConversation := conversation[mid:] - - newHistory := make([]providers.Message, 0) - newHistory = append(newHistory, history[0]) // System prompt - - // Add a note about compression - compressionNote := fmt.Sprintf("[System: Emergency compression dropped %d oldest messages due to context limit]", droppedCount) - // If there was an existing summary, we might lose it if it was in the dropped part (which is just messages). - // The summary is stored separately in session.Summary, so it persists! - // We just need to ensure the user knows there's a gap. - - // We only modify the messages list here - newHistory = append(newHistory, providers.Message{ - Role: "system", - Content: compressionNote, - }) - - newHistory = append(newHistory, keptConversation...) - newHistory = append(newHistory, history[len(history)-1]) // Last message - - // Update session - al.sessions.SetHistory(sessionKey, newHistory) - al.sessions.Save(sessionKey) - - logger.WarnCF("agent", "Forced compression executed", map[string]interface{}{ - "session_key": sessionKey, - "dropped_msgs": droppedCount, - "new_count": len(newHistory), - }) -} - // GetStartupInfo returns information about loaded tools and skills for logging. func (al *AgentLoop) GetStartupInfo() map[string]interface{} { info := make(map[string]interface{}) @@ -830,7 +642,7 @@ func formatMessagesForLog(messages []providers.Message) string { result += "[\n" for i, msg := range messages { result += fmt.Sprintf(" [%d] Role: %s\n", i, msg.Role) - if len(msg.ToolCalls) > 0 { + if msg.ToolCalls != nil && len(msg.ToolCalls) > 0 { result += " ToolCalls:\n" for _, tc := range msg.ToolCalls { result += fmt.Sprintf(" - ID: %s, Type: %s, Name: %s\n", tc.ID, tc.Type, tc.Name) @@ -897,7 +709,7 @@ func (al *AgentLoop) summarizeSession(sessionKey string) { continue } // Estimate tokens for this message - msgTokens := len(m.Content) / 2 // Use safer estimate here too (2.5 -> 2 for integer division safety) + msgTokens := len(m.Content) / 4 if msgTokens > maxMessageTokens { omitted = true continue @@ -968,96 +780,13 @@ func (al *AgentLoop) summarizeBatch(ctx context.Context, batch []providers.Messa } // estimateTokens estimates the number of tokens in a message list. -// Uses a safe heuristic of 2.5 characters per token to account for CJK and other -// overheads better than the previous 3 chars/token. +// Uses rune count instead of byte length so that CJK and other multi-byte +// characters are not over-counted (a Chinese character is 3 bytes but roughly +// one token). func (al *AgentLoop) estimateTokens(messages []providers.Message) int { - totalChars := 0 + total := 0 for _, m := range messages { - totalChars += utf8.RuneCountInString(m.Content) + total += utf8.RuneCountInString(m.Content) / 3 } - // 2.5 chars per token = totalChars * 2 / 5 - return totalChars * 2 / 5 -} - -func (al *AgentLoop) handleCommand(ctx context.Context, msg bus.InboundMessage) (string, bool) { - content := strings.TrimSpace(msg.Content) - if !strings.HasPrefix(content, "/") { - return "", false - } - - parts := strings.Fields(content) - if len(parts) == 0 { - return "", false - } - - cmd := parts[0] - args := parts[1:] - - switch cmd { - case "/show": - if len(args) < 1 { - return "Usage: /show [model|channel]", true - } - switch args[0] { - case "model": - return fmt.Sprintf("Current model: %s", al.model), true - case "channel": - return fmt.Sprintf("Current channel: %s", msg.Channel), true - default: - return fmt.Sprintf("Unknown show target: %s", args[0]), true - } - - case "/list": - if len(args) < 1 { - return "Usage: /list [models|channels]", true - } - switch args[0] { - case "models": - // TODO: Fetch available models dynamically if possible - return "Available models: glm-4.7, claude-3-5-sonnet, gpt-4o (configured in config.json/env)", true - case "channels": - if al.channelManager == nil { - return "Channel manager not initialized", true - } - channels := al.channelManager.GetEnabledChannels() - if len(channels) == 0 { - return "No channels enabled", true - } - return fmt.Sprintf("Enabled channels: %s", strings.Join(channels, ", ")), true - default: - return fmt.Sprintf("Unknown list target: %s", args[0]), true - } - - case "/switch": - if len(args) < 3 || args[1] != "to" { - return "Usage: /switch [model|channel] to ", true - } - target := args[0] - value := args[2] - - switch target { - case "model": - oldModel := al.model - al.model = value - return fmt.Sprintf("Switched model from %s to %s", oldModel, value), true - case "channel": - // This changes the 'default' channel for some operations, or effectively redirects output? - // For now, let's just validate if the channel exists - if al.channelManager == nil { - return "Channel manager not initialized", true - } - if _, exists := al.channelManager.GetChannel(value); !exists && value != "cli" { - return fmt.Sprintf("Channel '%s' not found or not enabled", value), true - } - - // If message came from CLI, maybe we want to redirect CLI output to this channel? - // That would require state persistence about "redirected channel" - // For now, just acknowledged. - return fmt.Sprintf("Switched target channel to %s (Note: this currently only validates existence)", value), true - default: - return fmt.Sprintf("Unknown switch target: %s", target), true - } - } - - return "", false + return total } diff --git a/pkg/channels/manager.go b/pkg/channels/manager.go index 7f6abc4cb..e1af01b52 100644 --- a/pkg/channels/manager.go +++ b/pkg/channels/manager.go @@ -48,7 +48,7 @@ func (m *Manager) initChannels() error { if m.config.Channels.Telegram.Enabled && m.config.Channels.Telegram.Token != "" { logger.DebugC("channels", "Attempting to initialize Telegram channel") - telegram, err := NewTelegramChannel(m.config, m.bus) + telegram, err := NewTelegramChannel(m.config.Channels.Telegram, m.bus) if err != nil { logger.ErrorCF("channels", "Failed to initialize Telegram channel", map[string]interface{}{ "error": err.Error(), @@ -176,6 +176,19 @@ func (m *Manager) initChannels() error { } } + if m.config.Channels.Pushover.Enabled && m.config.Channels.Pushover.AppToken != "" && m.config.Channels.Pushover.UserKey != "" { + logger.DebugC("channels", "Attempting to initialize Pushover channel") + pushover, err := NewPushoverChannel(m.config.Channels.Pushover, m.bus) + if err != nil { + logger.ErrorCF("channels", "Failed to initialize Pushover channel", map[string]interface{}{ + "error": err.Error(), + }) + } else { + m.channels["pushover"] = pushover + logger.InfoC("channels", "Pushover channel enabled successfully") + } + } + logger.InfoCF("channels", "Channel initialization completed", map[string]interface{}{ "enabled_channels": len(m.channels), }) diff --git a/pkg/channels/pushover.go b/pkg/channels/pushover.go new file mode 100644 index 000000000..839daf2df --- /dev/null +++ b/pkg/channels/pushover.go @@ -0,0 +1,88 @@ +package channels + +import ( + "context" + "fmt" + "net/http" + "net/url" + "strings" + + "github.com/sipeed/picoclaw/pkg/bus" + "github.com/sipeed/picoclaw/pkg/config" + "github.com/sipeed/picoclaw/pkg/logger" +) + +type PushoverChannel struct { + *BaseChannel + config config.PushoverConfig + client *http.Client +} + +func NewPushoverChannel(cfg config.PushoverConfig, bus *bus.MessageBus) (*PushoverChannel, error) { + base := NewBaseChannel("pushover", cfg, bus, nil) + + return &PushoverChannel{ + BaseChannel: base, + config: cfg, + client: &http.Client{}, + }, nil +} + +func (c *PushoverChannel) Name() string { + return "pushover" +} + +func (c *PushoverChannel) Start(ctx context.Context) error { + logger.InfoC("pushover", "Starting Pushover channel") + c.setRunning(true) + return nil +} + +func (c *PushoverChannel) Stop(ctx context.Context) error { + logger.InfoC("pushover", "Stopping Pushover channel") + c.setRunning(false) + return nil +} + +func (c *PushoverChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { + if !c.IsRunning() { + return fmt.Errorf("pushover channel not running") + } + + if c.config.AppToken == "" || c.config.UserKey == "" { + return fmt.Errorf("pushover app_token and user_key are required") + } + + data := url.Values{} + data.Set("token", c.config.AppToken) + data.Set("user", c.config.UserKey) + data.Set("message", msg.Content) + + // Truncate message if too long (Pushover limit is 1024 chars) + if len(msg.Content) > 1024 { + data.Set("message", msg.Content[:1021]+"...") + } + + req, err := http.NewRequestWithContext(ctx, "POST", "https://api.pushover.net/1/messages.json", strings.NewReader(data.Encode())) + if err != nil { + return fmt.Errorf("failed to create request: %w", err) + } + + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + + resp, err := c.client.Do(req) + if err != nil { + return fmt.Errorf("failed to send notification: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != 200 { + return fmt.Errorf("pushover API returned status %d", resp.StatusCode) + } + + logger.DebugCF("pushover", "Notification sent", map[string]any{ + "content_length": len(msg.Content), + }) + + return nil +} diff --git a/pkg/config/config.go b/pkg/config/config.go index d189ff00b..e8407120d 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -79,6 +79,7 @@ type ChannelsConfig struct { Slack SlackConfig `json:"slack"` LINE LINEConfig `json:"line"` OneBot OneBotConfig `json:"onebot"` + Pushover PushoverConfig `json:"pushover"` } type WhatsAppConfig struct { @@ -104,9 +105,10 @@ type FeishuConfig struct { } type DiscordConfig struct { - Enabled bool `json:"enabled" env:"PICOCLAW_CHANNELS_DISCORD_ENABLED"` - Token string `json:"token" env:"PICOCLAW_CHANNELS_DISCORD_TOKEN"` - AllowFrom FlexibleStringSlice `json:"allow_from" env:"PICOCLAW_CHANNELS_DISCORD_ALLOW_FROM"` + Enabled bool `json:"enabled" env:"PICOCLAW_CHANNELS_DISCORD_ENABLED"` + Token string `json:"token" env:"PICOCLAW_CHANNELS_DISCORD_TOKEN"` + AllowFrom FlexibleStringSlice `json:"allow_from" env:"PICOCLAW_CHANNELS_DISCORD_ALLOW_FROM"` + MentionOnly bool `json:"mention_only" env:"PICOCLAW_CHANNELS_DISCORD_MENTION_ONLY"` } type MaixCamConfig struct { @@ -156,6 +158,12 @@ type OneBotConfig struct { AllowFrom FlexibleStringSlice `json:"allow_from" env:"PICOCLAW_CHANNELS_ONEBOT_ALLOW_FROM"` } +type PushoverConfig struct { + Enabled bool `json:"enabled" env:"PICOCLAW_CHANNELS_PUSHOVER_ENABLED"` + AppToken string `json:"app_token" env:"PICOCLAW_CHANNELS_PUSHOVER_APP_TOKEN"` + UserKey string `json:"user_key" env:"PICOCLAW_CHANNELS_PUSHOVER_USER_KEY"` +} + type HeartbeatConfig struct { Enabled bool `json:"enabled" env:"PICOCLAW_HEARTBEAT_ENABLED"` Interval int `json:"interval" env:"PICOCLAW_HEARTBEAT_INTERVAL"` // minutes, min 5 @@ -175,7 +183,6 @@ type ProvidersConfig struct { VLLM ProviderConfig `json:"vllm"` Gemini ProviderConfig `json:"gemini"` Nvidia ProviderConfig `json:"nvidia"` - Ollama ProviderConfig `json:"ollama"` Moonshot ProviderConfig `json:"moonshot"` ShengSuanYun ProviderConfig `json:"shengsuanyun"` DeepSeek ProviderConfig `json:"deepseek"` @@ -370,7 +377,7 @@ func SaveConfig(path string, cfg *Config) error { return err } - return os.WriteFile(path, data, 0600) + return os.WriteFile(path, data, 0644) } func (c *Config) WorkspacePath() string { diff --git a/pkg/tools/pushover.go b/pkg/tools/pushover.go new file mode 100644 index 000000000..cc4e4bb7f --- /dev/null +++ b/pkg/tools/pushover.go @@ -0,0 +1,61 @@ +package tools + +import ( + "context" + "fmt" +) + +type PushoverTool struct { + pushoverCallback func(message string) error +} + +func NewPushoverTool() *PushoverTool { + return &PushoverTool{} +} + +func (t *PushoverTool) Name() string { + return "pushover" +} + +func (t *PushoverTool) Description() string { + return "Send a push notification to your phone via Pushover. Use this when you need to notify yourself of something important." +} + +func (t *PushoverTool) Parameters() map[string]interface{} { + return map[string]interface{}{ + "type": "object", + "properties": map[string]interface{}{ + "message": map[string]interface{}{ + "type": "string", + "description": "The notification message to send to your phone", + }, + }, + "required": []string{"message"}, + } +} + +func (t *PushoverTool) SetPushoverCallback(callback func(message string) error) { + t.pushoverCallback = callback +} + +func (t *PushoverTool) Execute(ctx context.Context, args map[string]interface{}) *ToolResult { + message, ok := args["message"].(string) + if !ok { + return &ToolResult{ForLLM: "message is required", IsError: true} + } + + if t.pushoverCallback == nil { + return &ToolResult{ForLLM: "Pushover not configured", IsError: true} + } + + if err := t.pushoverCallback(message); err != nil { + return &ToolResult{ + ForLLM: fmt.Sprintf("failed to send pushover notification: %v", err), + IsError: true, + } + } + + return &ToolResult{ + ForLLM: fmt.Sprintf("Push notification sent: %s", message), + } +} diff --git a/workspace/skills/pushover/SKILL.md b/workspace/skills/pushover/SKILL.md new file mode 100644 index 000000000..9bf843fdb --- /dev/null +++ b/workspace/skills/pushover/SKILL.md @@ -0,0 +1,43 @@ +--- +name: pushover +description: Send push notifications to your phone via Pushover. +metadata: {"picoclaw":{"emoji":"📱","requires":{"config":["channels.pushover"]}}} +--- + +# Pushover + +Send push notifications to your iPhone/Android via Pushover. + +## Usage + +Use the `pushover` tool to send notifications: + +```json +{ + "message": "Your notification message here" +} +``` + +## When to Use + +- Send heartbeat status notifications +- Alert yourself of important events +- Notify when long-running tasks complete + +## Setup + +Configure in `config.json`: + +```json +{ + "channels": { + "pushover": { + "enabled": true, + "app_token": "YOUR_APP_TOKEN", + "user_key": "YOUR_USER_KEY" + } + } +} +``` + +Get tokens from https://pushover.net/