404 lines
12 KiB
Go
404 lines
12 KiB
Go
// PicoClaw - Ultra-lightweight personal AI agent
|
|
// Inspired by and based on nanobot: https://github.com/HKUDS/nanobot
|
|
// License: MIT
|
|
//
|
|
// Copyright (c) 2026 PicoClaw contributors
|
|
|
|
package agent
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"sync"
|
|
"time"
|
|
|
|
"jane/pkg/bus"
|
|
"jane/pkg/channels"
|
|
"jane/pkg/commands"
|
|
"jane/pkg/config"
|
|
"jane/pkg/logger"
|
|
"jane/pkg/media"
|
|
"jane/pkg/providers"
|
|
"jane/pkg/skills"
|
|
"jane/pkg/state"
|
|
"jane/pkg/tools"
|
|
"jane/pkg/tools/web"
|
|
"jane/pkg/voice"
|
|
)
|
|
|
|
func NewAgentLoop(
|
|
cfg *config.Config,
|
|
msgBus *bus.MessageBus,
|
|
provider providers.LLMProvider,
|
|
) *AgentLoop {
|
|
registry := NewAgentRegistry(cfg, provider)
|
|
|
|
// Register shared tools to all agents
|
|
registerSharedTools(cfg, msgBus, registry, provider)
|
|
|
|
// Set up shared fallback chain
|
|
cooldown := providers.NewCooldownTracker()
|
|
fallbackChain := providers.NewFallbackChain(cooldown)
|
|
|
|
// Create state manager using default agent's workspace for channel recording
|
|
defaultAgent := registry.GetDefaultAgent()
|
|
var stateManager *state.Manager
|
|
if defaultAgent != nil {
|
|
stateManager = state.NewManager(defaultAgent.Workspace)
|
|
}
|
|
|
|
al := &AgentLoop{
|
|
bus: msgBus,
|
|
cfg: cfg,
|
|
registry: registry,
|
|
state: stateManager,
|
|
summarizing: sync.Map{},
|
|
summaryJobs: make(chan summaryJob, 100),
|
|
fallback: fallbackChain,
|
|
cmdRegistry: commands.NewRegistry(commands.BuiltinDefinitions()),
|
|
}
|
|
|
|
return al
|
|
}
|
|
|
|
// registerSharedTools registers tools that are shared across all agents (web, message, spawn).
|
|
func registerSharedTools(
|
|
cfg *config.Config,
|
|
msgBus *bus.MessageBus,
|
|
registry *AgentRegistry,
|
|
provider providers.LLMProvider,
|
|
) {
|
|
for _, agentID := range registry.ListAgentIDs() {
|
|
agent, ok := registry.GetAgent(agentID)
|
|
if !ok {
|
|
continue
|
|
}
|
|
|
|
if cfg.Tools.IsToolEnabled("web") {
|
|
searchTool, err := web.NewWebSearchTool(web.WebSearchToolOptions{
|
|
BraveAPIKeys: config.MergeAPIKeys(cfg.Tools.Web.Brave.APIKey, cfg.Tools.Web.Brave.APIKeys),
|
|
BraveMaxResults: cfg.Tools.Web.Brave.MaxResults,
|
|
BraveEnabled: cfg.Tools.Web.Brave.Enabled,
|
|
TavilyAPIKeys: config.MergeAPIKeys(cfg.Tools.Web.Tavily.APIKey, cfg.Tools.Web.Tavily.APIKeys),
|
|
TavilyBaseURL: cfg.Tools.Web.Tavily.BaseURL,
|
|
TavilyMaxResults: cfg.Tools.Web.Tavily.MaxResults,
|
|
TavilyEnabled: cfg.Tools.Web.Tavily.Enabled,
|
|
DuckDuckGoMaxResults: cfg.Tools.Web.DuckDuckGo.MaxResults,
|
|
DuckDuckGoEnabled: cfg.Tools.Web.DuckDuckGo.Enabled,
|
|
PerplexityAPIKeys: config.MergeAPIKeys(
|
|
cfg.Tools.Web.Perplexity.APIKey,
|
|
cfg.Tools.Web.Perplexity.APIKeys,
|
|
),
|
|
PerplexityMaxResults: cfg.Tools.Web.Perplexity.MaxResults,
|
|
PerplexityEnabled: cfg.Tools.Web.Perplexity.Enabled,
|
|
SearXNGBaseURL: cfg.Tools.Web.SearXNG.BaseURL,
|
|
SearXNGMaxResults: cfg.Tools.Web.SearXNG.MaxResults,
|
|
SearXNGEnabled: cfg.Tools.Web.SearXNG.Enabled,
|
|
GLMSearchAPIKey: cfg.Tools.Web.GLMSearch.APIKey,
|
|
GLMSearchBaseURL: cfg.Tools.Web.GLMSearch.BaseURL,
|
|
GLMSearchEngine: cfg.Tools.Web.GLMSearch.SearchEngine,
|
|
GLMSearchMaxResults: cfg.Tools.Web.GLMSearch.MaxResults,
|
|
GLMSearchEnabled: cfg.Tools.Web.GLMSearch.Enabled,
|
|
Proxy: cfg.Tools.Web.Proxy,
|
|
})
|
|
if err != nil {
|
|
logger.ErrorCF("agent", "Failed to create web search tool", map[string]any{"error": err.Error()})
|
|
} else if searchTool != nil {
|
|
agent.Tools.Register(searchTool)
|
|
}
|
|
}
|
|
if cfg.Tools.IsToolEnabled("web_fetch") {
|
|
fetchTool, err := web.NewWebFetchToolWithProxy(50000, cfg.Tools.Web.Proxy, cfg.Tools.Web.FetchLimitBytes)
|
|
if err != nil {
|
|
logger.ErrorCF("agent", "Failed to create web fetch tool", map[string]any{"error": err.Error()})
|
|
} else {
|
|
agent.Tools.Register(fetchTool)
|
|
}
|
|
}
|
|
|
|
if cfg.Tools.IsToolEnabled("browser_action") {
|
|
browserActionTool := tools.NewBrowserActionTool()
|
|
agent.Tools.Register(browserActionTool)
|
|
}
|
|
|
|
if cfg.Tools.IsToolEnabled("go_eval") {
|
|
goEvalTool := tools.NewGoEvalTool(agent.Workspace)
|
|
agent.Tools.Register(goEvalTool)
|
|
}
|
|
|
|
if cfg.Tools.IsToolEnabled("mcp2cli") {
|
|
// We can initialize an empty MCP manager for mcp2cli if it's the only one,
|
|
// or share the existing MCP manager if one exists
|
|
mcp2CliTool := tools.NewMCP2CliTool(nil) // It will init its own manager or use a global one later if needed
|
|
agent.Tools.Register(mcp2CliTool)
|
|
}
|
|
|
|
// Hardware tools (I2C, SPI) - Linux only, returns error on other platforms
|
|
if cfg.Tools.IsToolEnabled("i2c") {
|
|
agent.Tools.Register(tools.NewI2CTool())
|
|
}
|
|
if cfg.Tools.IsToolEnabled("spi") {
|
|
agent.Tools.Register(tools.NewSPITool())
|
|
}
|
|
|
|
// Message tool
|
|
if cfg.Tools.IsToolEnabled("message") {
|
|
messageTool := tools.NewMessageTool()
|
|
messageTool.SetSendCallback(func(channel, chatID, content string) error {
|
|
pubCtx, pubCancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
defer pubCancel()
|
|
return msgBus.PublishOutbound(pubCtx, bus.OutboundMessage{
|
|
Channel: channel,
|
|
ChatID: chatID,
|
|
Content: content,
|
|
})
|
|
})
|
|
agent.Tools.Register(messageTool)
|
|
}
|
|
|
|
// Send file tool (outbound media via MediaStore — store injected later by SetMediaStore)
|
|
if cfg.Tools.IsToolEnabled("send_file") {
|
|
sendFileTool := tools.NewSendFileTool(
|
|
agent.Workspace,
|
|
cfg.Agents.Defaults.RestrictToWorkspace,
|
|
cfg.Agents.Defaults.GetMaxMediaSize(),
|
|
nil,
|
|
)
|
|
agent.Tools.Register(sendFileTool)
|
|
}
|
|
|
|
// Skill discovery and installation tools
|
|
skills_enabled := cfg.Tools.IsToolEnabled("skills")
|
|
find_skills_enable := cfg.Tools.IsToolEnabled("find_skills")
|
|
install_skills_enable := cfg.Tools.IsToolEnabled("install_skill")
|
|
if skills_enabled && (find_skills_enable || install_skills_enable) {
|
|
registryMgr := skills.NewRegistryManagerFromConfig(skills.RegistryConfig{
|
|
MaxConcurrentSearches: cfg.Tools.Skills.MaxConcurrentSearches,
|
|
ClawHub: skills.ClawHubConfig(cfg.Tools.Skills.Registries.ClawHub),
|
|
})
|
|
|
|
if find_skills_enable {
|
|
searchCache := skills.NewSearchCache(
|
|
cfg.Tools.Skills.SearchCache.MaxSize,
|
|
time.Duration(cfg.Tools.Skills.SearchCache.TTLSeconds)*time.Second,
|
|
)
|
|
agent.Tools.Register(tools.NewFindSkillsTool(registryMgr, searchCache))
|
|
}
|
|
|
|
if install_skills_enable {
|
|
agent.Tools.Register(tools.NewInstallSkillTool(registryMgr, agent.Workspace))
|
|
}
|
|
}
|
|
|
|
// Spawn tool with allowlist checker
|
|
if cfg.Tools.IsToolEnabled("spawn") {
|
|
if cfg.Tools.IsToolEnabled("subagent") {
|
|
subagentManager := tools.NewSubagentManager(provider, agent.Model, agent.Workspace)
|
|
subagentManager.SetLLMOptions(agent.MaxTokens, agent.Temperature)
|
|
spawnTool := tools.NewSpawnTool(subagentManager)
|
|
currentAgentID := agentID
|
|
spawnTool.SetAllowlistChecker(func(targetAgentID string) bool {
|
|
return registry.CanSpawnSubagent(currentAgentID, targetAgentID)
|
|
})
|
|
agent.Tools.Register(spawnTool)
|
|
} else {
|
|
logger.WarnCF("agent", "spawn tool requires subagent to be enabled", nil)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func (al *AgentLoop) Run(ctx context.Context) error {
|
|
al.running.Store(true)
|
|
if err := al.ensureMCPInitialized(ctx); err != nil {
|
|
return err
|
|
}
|
|
|
|
// Start background summarization worker
|
|
al.wg.Add(1)
|
|
go func() {
|
|
defer al.wg.Done()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case job, ok := <-al.summaryJobs:
|
|
if !ok {
|
|
return
|
|
}
|
|
logger.Debug("Memory threshold reached. Optimizing conversation history...")
|
|
al.summarizeSession(job.agent, job.sessionKey)
|
|
al.summarizing.Delete(job.agent.ID + ":" + job.sessionKey)
|
|
}
|
|
}
|
|
}()
|
|
|
|
for al.running.Load() {
|
|
select {
|
|
case <-ctx.Done():
|
|
al.wg.Wait()
|
|
return nil
|
|
default:
|
|
msg, ok := al.bus.ConsumeInbound(ctx)
|
|
if !ok {
|
|
continue
|
|
}
|
|
|
|
al.wg.Add(1)
|
|
// Process message
|
|
go func(msg bus.InboundMessage) {
|
|
defer al.wg.Done()
|
|
// TODO: Re-enable media cleanup after inbound media is properly consumed by the agent.
|
|
// Currently disabled because files are deleted before the LLM can access their content.
|
|
// defer func() {
|
|
// if al.mediaStore != nil && msg.MediaScope != "" {
|
|
// if releaseErr := al.mediaStore.ReleaseAll(msg.MediaScope); releaseErr != nil {
|
|
// logger.WarnCF("agent", "Failed to release media", map[string]any{
|
|
// "scope": msg.MediaScope,
|
|
// "error": releaseErr.Error(),
|
|
// })
|
|
// }
|
|
// }
|
|
// }()
|
|
|
|
response, err := al.processMessage(ctx, msg)
|
|
if err != nil {
|
|
response = fmt.Sprintf("Error processing message: %v", err)
|
|
}
|
|
|
|
if response != "" {
|
|
// Check if the message tool already sent a response during this round.
|
|
// If so, skip publishing to avoid duplicate messages to the user.
|
|
// Use default agent's tools to check (message tool is shared).
|
|
alreadySent := false
|
|
defaultAgent := al.registry.GetDefaultAgent()
|
|
if defaultAgent != nil {
|
|
if tool, ok := defaultAgent.Tools.Get("message"); ok {
|
|
if mt, ok := tool.(*tools.MessageTool); ok {
|
|
alreadySent = mt.HasSentInRound()
|
|
}
|
|
}
|
|
}
|
|
|
|
if !alreadySent {
|
|
al.bus.PublishOutbound(ctx, bus.OutboundMessage{
|
|
Channel: msg.Channel,
|
|
ChatID: msg.ChatID,
|
|
Content: response,
|
|
})
|
|
logger.InfoCF("agent", "Published outbound response",
|
|
map[string]any{
|
|
"channel": msg.Channel,
|
|
"chat_id": msg.ChatID,
|
|
"content_len": len(response),
|
|
})
|
|
} else {
|
|
logger.DebugCF(
|
|
"agent",
|
|
"Skipped outbound (message tool already sent)",
|
|
map[string]any{"channel": msg.Channel},
|
|
)
|
|
}
|
|
}
|
|
}(msg)
|
|
}
|
|
}
|
|
|
|
al.wg.Wait()
|
|
return nil
|
|
}
|
|
|
|
func (al *AgentLoop) Stop() {
|
|
al.running.Store(false)
|
|
}
|
|
|
|
// Close releases resources held by agent session stores. Call after Stop.
|
|
func (al *AgentLoop) Close() {
|
|
mcpManager := al.mcp.takeManager()
|
|
|
|
if mcpManager != nil {
|
|
if err := mcpManager.Close(); err != nil {
|
|
logger.ErrorCF("agent", "Failed to close MCP manager",
|
|
map[string]any{
|
|
"error": err.Error(),
|
|
})
|
|
}
|
|
}
|
|
|
|
al.registry.Close()
|
|
}
|
|
|
|
func (al *AgentLoop) RegisterTool(tool tools.Tool) {
|
|
for _, agentID := range al.registry.ListAgentIDs() {
|
|
if agent, ok := al.registry.GetAgent(agentID); ok {
|
|
agent.Tools.Register(tool)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (al *AgentLoop) SetChannelManager(cm *channels.Manager) {
|
|
al.channelManager = cm
|
|
}
|
|
|
|
// SetMediaStore injects a MediaStore for media lifecycle management.
|
|
func (al *AgentLoop) SetMediaStore(s media.MediaStore) {
|
|
al.mediaStore = s
|
|
|
|
// Propagate store to send_file tools in all agents.
|
|
al.registry.ForEachTool("send_file", func(t tools.Tool) {
|
|
if sf, ok := t.(*tools.SendFileTool); ok {
|
|
sf.SetMediaStore(s)
|
|
}
|
|
})
|
|
}
|
|
|
|
// SetTranscriber injects a voice transcriber for agent-level audio transcription.
|
|
func (al *AgentLoop) SetTranscriber(t voice.Transcriber) {
|
|
al.transcriber = t
|
|
}
|
|
|
|
// 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 {
|
|
if al.state == nil {
|
|
return nil
|
|
}
|
|
return al.state.SetLastChannel(channel)
|
|
}
|
|
|
|
// RecordLastChatID records the last active chat ID for this workspace.
|
|
// This uses the atomic state save mechanism to prevent data loss on crash.
|
|
func (al *AgentLoop) RecordLastChatID(chatID string) error {
|
|
if al.state == nil {
|
|
return nil
|
|
}
|
|
return al.state.SetLastChatID(chatID)
|
|
}
|
|
|
|
// GetStartupInfo returns information about loaded tools and skills for logging.
|
|
func (al *AgentLoop) GetStartupInfo() map[string]any {
|
|
info := make(map[string]any)
|
|
|
|
agent := al.registry.GetDefaultAgent()
|
|
if agent == nil {
|
|
return info
|
|
}
|
|
|
|
// Tools info
|
|
toolsList := agent.Tools.List()
|
|
info["tools"] = map[string]any{
|
|
"count": len(toolsList),
|
|
"names": toolsList,
|
|
}
|
|
|
|
// Skills info
|
|
info["skills"] = agent.ContextBuilder.GetSkillsInfo()
|
|
|
|
// Agents info
|
|
info["agents"] = map[string]any{
|
|
"count": len(al.registry.ListAgentIDs()),
|
|
"ids": al.registry.ListAgentIDs(),
|
|
}
|
|
|
|
return info
|
|
}
|