opencode/picoclaw
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add Pushover notification channel and tool

Browse source
This commit is contained in:
Vernon Stinebaker 2026-02-16 23:38:07 +08:00
parent 13e4028d42
commit fead0e8f4e
6 changed files with 245 additions and 304 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

325
pkg/agent/loop.go
View file

@ -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 <name>", 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
}

15
pkg/channels/manager.go
View file

@ -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),
})

88
pkg/channels/pushover.go Normal file
View file

@ -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
}

17
pkg/config/config.go
View file

@ -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 {

61
pkg/tools/pushover.go Normal file
View file

@ -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),
}
}

43
workspace/skills/pushover/SKILL.md Normal file
View file

@ -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/