From 3cceb99d22bf2b4a44da9fef4fbd82ce7a07b6f9 Mon Sep 17 00:00:00 2001 From: xj Date: Sat, 28 Feb 2026 17:38:53 -0800 Subject: [PATCH] feat(agent): wire plugin bootstrap into agent and gateway --- cmd/picoclaw/internal/agent/helpers.go | 19 + cmd/picoclaw/internal/gateway/helpers.go | 18 + pkg/agent/loop.go | 523 ++++++++++++----------- pkg/agent/loop_test.go | 84 ++++ pkg/agent/plugin_test.go | 416 ++++++++++++++++++ 5 files changed, 821 insertions(+), 239 deletions(-) create mode 100644 pkg/agent/plugin_test.go diff --git a/cmd/picoclaw/internal/agent/helpers.go b/cmd/picoclaw/internal/agent/helpers.go index f754abc65..f89c23795 100644 --- a/cmd/picoclaw/internal/agent/helpers.go +++ b/cmd/picoclaw/internal/agent/helpers.go @@ -12,6 +12,7 @@ import ( "github.com/chzyer/readline" "github.com/sipeed/picoclaw/cmd/picoclaw/internal" + "github.com/sipeed/picoclaw/cmd/picoclaw/internal/pluginruntime" "github.com/sipeed/picoclaw/pkg/agent" "github.com/sipeed/picoclaw/pkg/bus" "github.com/sipeed/picoclaw/pkg/logger" @@ -51,6 +52,24 @@ func agentCmd(message, sessionKey, model string, debug bool) error { defer msgBus.Close() agentLoop := agent.NewAgentLoop(cfg, msgBus, provider) + pluginsToEnable, pluginSummary, err := pluginruntime.ResolveConfiguredPlugins(cfg) + if err != nil { + return fmt.Errorf("error resolving configured plugins: %w", err) + } + if len(pluginsToEnable) > 0 { + if err := agentLoop.EnablePlugins(pluginsToEnable...); err != nil { + return fmt.Errorf("error enabling plugins: %w", err) + } + } + logger.InfoCF("agent", "Plugin selection resolved", + map[string]any{ + "plugins_enabled": pluginSummary.Enabled, + "plugins_disabled": pluginSummary.Disabled, + "plugins_unknown_enabled": pluginSummary.UnknownEnabled, + "plugins_unknown_disabled": pluginSummary.UnknownDisabled, + "plugins_warnings": pluginSummary.Warnings, + }) + // Print agent startup info (only for interactive mode) startupInfo := agentLoop.GetStartupInfo() logger.InfoCF("agent", "Agent initialized", diff --git a/cmd/picoclaw/internal/gateway/helpers.go b/cmd/picoclaw/internal/gateway/helpers.go index 747f7d44e..154b221fb 100644 --- a/cmd/picoclaw/internal/gateway/helpers.go +++ b/cmd/picoclaw/internal/gateway/helpers.go @@ -10,6 +10,7 @@ import ( "time" "github.com/sipeed/picoclaw/cmd/picoclaw/internal" + "github.com/sipeed/picoclaw/cmd/picoclaw/internal/pluginruntime" "github.com/sipeed/picoclaw/pkg/agent" "github.com/sipeed/picoclaw/pkg/bus" "github.com/sipeed/picoclaw/pkg/channels" @@ -61,6 +62,23 @@ func gatewayCmd(debug bool) error { msgBus := bus.NewMessageBus() agentLoop := agent.NewAgentLoop(cfg, msgBus, provider) + pluginsToEnable, pluginSummary, err := pluginruntime.ResolveConfiguredPlugins(cfg) + if err != nil { + return fmt.Errorf("error resolving configured plugins: %w", err) + } + if len(pluginsToEnable) > 0 { + if err := agentLoop.EnablePlugins(pluginsToEnable...); err != nil { + return fmt.Errorf("error enabling plugins: %w", err) + } + } + logger.InfoCF("agent", "Plugin selection resolved", + map[string]any{ + "plugins_enabled": pluginSummary.Enabled, + "plugins_disabled": pluginSummary.Disabled, + "plugins_unknown_enabled": pluginSummary.UnknownEnabled, + "plugins_unknown_disabled": pluginSummary.UnknownDisabled, + "plugins_warnings": pluginSummary.Warnings, + }) // Print agent startup info fmt.Println("\nšŸ“¦ Agent Status:") diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index 8fd7328d1..a4d543c73 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -9,9 +9,7 @@ package agent import ( "context" "encoding/json" - "errors" "fmt" - "path/filepath" "strings" "sync" "sync/atomic" @@ -22,8 +20,9 @@ import ( "github.com/sipeed/picoclaw/pkg/channels" "github.com/sipeed/picoclaw/pkg/config" "github.com/sipeed/picoclaw/pkg/constants" + "github.com/sipeed/picoclaw/pkg/hooks" "github.com/sipeed/picoclaw/pkg/logger" - "github.com/sipeed/picoclaw/pkg/media" + "github.com/sipeed/picoclaw/pkg/plugin" "github.com/sipeed/picoclaw/pkg/providers" "github.com/sipeed/picoclaw/pkg/routing" "github.com/sipeed/picoclaw/pkg/skills" @@ -41,7 +40,8 @@ type AgentLoop struct { summarizing sync.Map fallback *providers.FallbackChain channelManager *channels.Manager - mediaStore media.MediaStore + hooks *hooks.HookRegistry + pluginManager *plugin.Manager } // processOptions configures how a message is processed @@ -56,8 +56,6 @@ type processOptions struct { NoHistory bool // If true, don't load session history (for heartbeat) } -const defaultResponse = "I've completed processing but have no response to give. Increase `max_tool_iterations` in config.json." - func NewAgentLoop(cfg *config.Config, msgBus *bus.MessageBus, provider providers.LLMProvider) *AgentLoop { registry := NewAgentRegistry(cfg, provider) @@ -124,14 +122,13 @@ func registerSharedTools( // Message tool 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{ + messageTool.SetSendCallback(func(_ context.Context, channel, chatID, content string) error { + msgBus.PublishOutbound(bus.OutboundMessage{ Channel: channel, ChatID: chatID, Content: content, }) + return nil }) agent.Tools.Register(messageTool) @@ -172,61 +169,33 @@ func (al *AgentLoop) Run(ctx context.Context) error { continue } - // Process message - func() { - // 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) + } - 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 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}, - ) - } } - }() + + if !alreadySent { + al.sendOutbound(ctx, bus.OutboundMessage{ + Channel: msg.Channel, + ChatID: msg.ChatID, + Content: response, + }) + } + } } } @@ -249,39 +218,98 @@ 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 +// SetHooks installs a hook registry. Must be called before Run starts. +func (al *AgentLoop) SetHooks(h *hooks.HookRegistry) error { + if al.running.Load() { + return fmt.Errorf("SetHooks must be called before Run starts") + } + al.hooks = h + + // Rewire MessageTool callbacks to route through sendOutbound for hook interception. + for _, agentID := range al.registry.ListAgentIDs() { + if agent, ok := al.registry.GetAgent(agentID); ok { + if tool, ok := agent.Tools.Get("message"); ok { + if mt, ok := tool.(*tools.MessageTool); ok { + if h == nil { + mt.SetSendCallback(func(_ context.Context, channel, chatID, content string) error { + al.bus.PublishOutbound(bus.OutboundMessage{ + Channel: channel, + ChatID: chatID, + Content: content, + }) + return nil + }) + continue + } + mt.SetSendCallback(func(ctx context.Context, channel, chatID, content string) error { + if sent, reason := al.sendOutbound(ctx, bus.OutboundMessage{ + Channel: channel, + ChatID: chatID, + Content: content, + }); !sent { + if strings.TrimSpace(reason) == "" { + reason = "unspecified" + } + return fmt.Errorf("message canceled by hook: %s", reason) + } + return nil + }) + } + } + } + } + return nil } -// inferMediaType determines the media type ("image", "audio", "video", "file") -// from a filename and MIME content type. -func inferMediaType(filename, contentType string) string { - ct := strings.ToLower(contentType) - fn := strings.ToLower(filename) +// SetPluginManager installs a plugin manager and routes its hook registry into the loop. +// Must be called before Run starts. +func (al *AgentLoop) SetPluginManager(pm *plugin.Manager) error { + if pm == nil { + if err := al.SetHooks(nil); err != nil { + return err + } + al.pluginManager = nil + return nil + } + if err := al.SetHooks(pm.HookRegistry()); err != nil { + return err + } + al.pluginManager = pm + return nil +} - if strings.HasPrefix(ct, "image/") { - return "image" - } - if strings.HasPrefix(ct, "audio/") || ct == "application/ogg" { - return "audio" - } - if strings.HasPrefix(ct, "video/") { - return "video" +// EnablePlugins is a convenience helper to build and install a plugin manager. +func (al *AgentLoop) EnablePlugins(plugins ...plugin.Plugin) error { + pm := plugin.NewManager() + if err := pm.RegisterAll(plugins...); err != nil { + return err } + return al.SetPluginManager(pm) +} - // Fallback: infer from extension - ext := filepath.Ext(fn) - switch ext { - case ".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp", ".svg": - return "image" - case ".mp3", ".wav", ".ogg", ".m4a", ".flac", ".aac", ".wma", ".opus": - return "audio" - case ".mp4", ".avi", ".mov", ".webm", ".mkv": - return "video" +// sendOutbound wraps bus.PublishOutbound with the message_sending hook. +// Returns whether the message was sent and, if canceled, the cancel reason. +func (al *AgentLoop) sendOutbound(ctx context.Context, msg bus.OutboundMessage) (bool, string) { + if al.hooks != nil { + event := &hooks.MessageSendingEvent{Channel: msg.Channel, ChatID: msg.ChatID, Content: msg.Content} + al.hooks.TriggerMessageSending(ctx, event) + if event.Cancel { + reason := event.CancelReason + if reason == "" { + reason = "unspecified" + } + logger.WarnCF("hooks", "Outbound message canceled by hook", + map[string]any{ + "channel": msg.Channel, + "chat_id": msg.ChatID, + "reason": reason, + }) + return false, reason + } + msg.Content = event.Content } - - return "file" + al.bus.PublishOutbound(msg) + return true, "" } // RecordLastChannel records the last active channel for this workspace. @@ -325,15 +353,12 @@ func (al *AgentLoop) ProcessDirectWithChannel( // Each heartbeat is independent and doesn't accumulate context. func (al *AgentLoop) ProcessHeartbeat(ctx context.Context, content, channel, chatID string) (string, error) { agent := al.registry.GetDefaultAgent() - if agent == nil { - return "", fmt.Errorf("no default agent for heartbeat") - } return al.runAgentLoop(ctx, agent, processOptions{ SessionKey: "heartbeat", Channel: channel, ChatID: chatID, UserMessage: content, - DefaultResponse: defaultResponse, + DefaultResponse: "I've completed processing but have no response to give.", EnableSummary: false, SendResponse: false, NoHistory: true, // Don't load session history for heartbeat @@ -356,6 +381,18 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage) "session_key": msg.SessionKey, }) + // Fire message_received hook + if al.hooks != nil { + al.hooks.TriggerMessageReceived(ctx, &hooks.MessageReceivedEvent{ + Channel: msg.Channel, + SenderID: msg.SenderID, + ChatID: msg.ChatID, + Content: msg.Content, + Media: msg.Media, + Metadata: msg.Metadata, + }) + } + // Route system messages to processSystemMessage if msg.Channel == "system" { return al.processSystemMessage(ctx, msg) @@ -380,16 +417,6 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage) if !ok { agent = al.registry.GetDefaultAgent() } - if agent == nil { - return "", fmt.Errorf("no agent available for route (agent_id=%s)", route.AgentID) - } - - // Reset message-tool state for this round so we don't skip publishing due to a previous round. - if tool, ok := agent.Tools.Get("message"); ok { - if mt, ok := tool.(tools.ContextualTool); ok { - mt.SetContext(msg.Channel, msg.ChatID) - } - } // Use routed session key, but honor pre-set agent-scoped keys (for ProcessDirect/cron) sessionKey := route.SessionKey @@ -409,7 +436,7 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage) Channel: msg.Channel, ChatID: msg.ChatID, UserMessage: msg.Content, - DefaultResponse: defaultResponse, + DefaultResponse: "I've completed processing but have no response to give.", EnableSummary: true, SendResponse: false, }) @@ -456,9 +483,6 @@ func (al *AgentLoop) processSystemMessage(ctx context.Context, msg bus.InboundMe // Use default agent for system messages agent := al.registry.GetDefaultAgent() - if agent == nil { - return "", fmt.Errorf("no default agent for system message") - } // Use the origin session for context sessionKey := routing.BuildAgentMainSessionKey(agent.ID) @@ -490,6 +514,18 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, agent *AgentInstance, opt // 1. Update tool contexts al.updateToolContexts(agent, opts.Channel, opts.ChatID) + // Fire session hooks + if al.hooks != nil { + sessionEvt := &hooks.SessionEvent{ + AgentID: agent.ID, + SessionKey: opts.SessionKey, + Channel: opts.Channel, + ChatID: opts.ChatID, + } + al.hooks.TriggerSessionStart(ctx, sessionEvt) + defer al.hooks.TriggerSessionEnd(ctx, sessionEvt) + } + // 2. Build messages (skip history for heartbeat) var history []providers.Message var summary string @@ -529,12 +565,12 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, agent *AgentInstance, opt // 7. Optional: summarization if opts.EnableSummary { - al.maybeSummarize(agent, opts.SessionKey, opts.Channel, opts.ChatID) + al.maybeSummarize(ctx, agent, opts.SessionKey, opts.Channel, opts.ChatID) } // 8. Optional: send response via bus if opts.SendResponse { - al.bus.PublishOutbound(ctx, bus.OutboundMessage{ + al.sendOutbound(ctx, bus.OutboundMessage{ Channel: opts.Channel, ChatID: opts.ChatID, Content: finalContent, @@ -554,59 +590,6 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, agent *AgentInstance, opt return finalContent, nil } -func (al *AgentLoop) targetReasoningChannelID(channelName string) (chatID string) { - if al.channelManager == nil { - return "" - } - if ch, ok := al.channelManager.GetChannel(channelName); ok { - return ch.ReasoningChannelID() - } - return "" -} - -func (al *AgentLoop) handleReasoning(ctx context.Context, reasoningContent, channelName, channelID string) { - if reasoningContent == "" || channelName == "" || channelID == "" { - return - } - - // Check context cancellation before attempting to publish, - // since PublishOutbound's select may race between send and ctx.Done(). - if ctx.Err() != nil { - return - } - - // Use a short timeout so the goroutine does not block indefinitely when - // the outbound bus is full. Reasoning output is best-effort; dropping it - // is acceptable to avoid goroutine accumulation. - pubCtx, pubCancel := context.WithTimeout(ctx, 5*time.Second) - defer pubCancel() - - if err := al.bus.PublishOutbound(pubCtx, bus.OutboundMessage{ - Channel: channelName, - ChatID: channelID, - Content: reasoningContent, - }); err != nil { - // Treat context.DeadlineExceeded / context.Canceled as expected - // (bus full under load, or parent canceled). Check the error - // itself rather than ctx.Err(), because pubCtx may time out - // (5 s) while the parent ctx is still active. - // Also treat ErrBusClosed as expected ā€” it occurs during normal - // shutdown when the bus is closed before all goroutines finish. - if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) || - errors.Is(err, bus.ErrBusClosed) { - logger.DebugCF("agent", "Reasoning publish skipped (timeout/cancel)", map[string]any{ - "channel": channelName, - "error": err.Error(), - }) - } else { - logger.WarnCF("agent", "Failed to publish reasoning (best-effort)", map[string]any{ - "channel": channelName, - "error": err.Error(), - }) - } - } -} - // runLLMIteration executes the LLM call loop with tool handling. func (al *AgentLoop) runLLMIteration( ctx context.Context, @@ -684,43 +667,29 @@ func (al *AgentLoop) runLLMIteration( } // Retry loop for context/token errors + llmStart := time.Now() maxRetries := 2 for retry := 0; retry <= maxRetries; retry++ { + // Fire llm_input hook (re-fires after compression so hooks see actual messages) + if al.hooks != nil { + al.hooks.TriggerLLMInput(ctx, &hooks.LLMInputEvent{ + AgentID: agent.ID, + Model: agent.Model, + Messages: messages, + Tools: providerToolDefs, + Iteration: iteration, + }) + } response, err = callLLM() if err == nil { break } errMsg := strings.ToLower(err.Error()) - - // Check if this is a network/HTTP timeout ā€” not a context window error. - isTimeoutError := errors.Is(err, context.DeadlineExceeded) || - strings.Contains(errMsg, "deadline exceeded") || - strings.Contains(errMsg, "client.timeout") || - strings.Contains(errMsg, "timed out") || - strings.Contains(errMsg, "timeout exceeded") - - // Detect real context window / token limit errors, excluding network timeouts. - isContextError := !isTimeoutError && (strings.Contains(errMsg, "context_length_exceeded") || - strings.Contains(errMsg, "context window") || - strings.Contains(errMsg, "maximum context length") || - strings.Contains(errMsg, "token limit") || - strings.Contains(errMsg, "too many tokens") || - strings.Contains(errMsg, "max_tokens") || + isContextError := strings.Contains(errMsg, "token") || + strings.Contains(errMsg, "context") || strings.Contains(errMsg, "invalidparameter") || - strings.Contains(errMsg, "prompt is too long") || - strings.Contains(errMsg, "request too large")) - - if isTimeoutError && retry < maxRetries { - backoff := time.Duration(retry+1) * 5 * time.Second - logger.WarnCF("agent", "Timeout error, retrying after backoff", map[string]any{ - "error": err.Error(), - "retry": retry, - "backoff": backoff.String(), - }) - time.Sleep(backoff) - continue - } + strings.Contains(errMsg, "length") if isContextError && retry < maxRetries { logger.WarnCF("agent", "Context window error detected, attempting compression", map[string]any{ @@ -729,7 +698,7 @@ func (al *AgentLoop) runLLMIteration( }) if retry == 0 && !constants.IsInternalChannel(opts.Channel) { - al.bus.PublishOutbound(ctx, bus.OutboundMessage{ + al.sendOutbound(ctx, bus.OutboundMessage{ Channel: opts.Channel, ChatID: opts.ChatID, Content: "Context window exceeded. Compressing history and retrying...", @@ -748,6 +717,8 @@ func (al *AgentLoop) runLLMIteration( break } + llmDuration := time.Since(llmStart) + if err != nil { logger.ErrorCF("agent", "LLM call failed", map[string]any{ @@ -758,18 +729,18 @@ func (al *AgentLoop) runLLMIteration( return "", iteration, fmt.Errorf("LLM call failed after retries: %w", err) } - go al.handleReasoning(ctx, response.Reasoning, opts.Channel, al.targetReasoningChannelID(opts.Channel)) - - logger.DebugCF("agent", "LLM response", - map[string]any{ - "agent_id": agent.ID, - "iteration": iteration, - "content_chars": len(response.Content), - "tool_calls": len(response.ToolCalls), - "reasoning": response.Reasoning, - "target_channel": al.targetReasoningChannelID(opts.Channel), - "channel": opts.Channel, + // Fire llm_output hook + if al.hooks != nil { + al.hooks.TriggerLLMOutput(ctx, &hooks.LLMOutputEvent{ + AgentID: agent.ID, + Model: agent.Model, + Content: response.Content, + ToolCalls: response.ToolCalls, + Iteration: iteration, + Duration: llmDuration, }) + } + // Check if no tool calls - we're done if len(response.ToolCalls) == 0 { finalContent = response.Content @@ -832,9 +803,14 @@ func (al *AgentLoop) runLLMIteration( // Save assistant message with tool calls to session agent.Sessions.AddFullMessage(opts.SessionKey, assistantMsg) + assistantMsgIndex := len(messages) - 1 + assistantSessionIndex := -1 + if history := agent.Sessions.GetHistory(opts.SessionKey); len(history) > 0 { + assistantSessionIndex = len(history) - 1 + } // Execute tool calls - for _, tc := range normalizedToolCalls { + for tcIdx, tc := range normalizedToolCalls { argsJSON, _ := json.Marshal(tc.Arguments) argsPreview := utils.Truncate(string(argsJSON), 200) logger.InfoCF("agent", fmt.Sprintf("Tool call: %s(%s)", tc.Name, argsPreview), @@ -860,18 +836,74 @@ func (al *AgentLoop) runLLMIteration( } } - toolResult := agent.Tools.ExecuteWithContext( - ctx, - tc.Name, - tc.Arguments, - opts.Channel, - opts.ChatID, - asyncCallback, - ) + // Fire before_tool_call hook + var toolResult *tools.ToolResult + toolCanceled := false + if al.hooks != nil { + args := tc.Arguments + if args == nil { + args = make(map[string]any) + } + btcEvent := &hooks.BeforeToolCallEvent{ + ToolName: tc.Name, + Args: args, + Channel: opts.Channel, + ChatID: opts.ChatID, + } + al.hooks.TriggerBeforeToolCall(ctx, btcEvent) + if btcEvent.Cancel { + toolCanceled = true + reason := btcEvent.CancelReason + if strings.TrimSpace(reason) == "" { + reason = fmt.Sprintf("tool call %q was canceled by before_tool_call hook", tc.Name) + } + toolResult = tools.ErrorResult(reason) + } + tc.Arguments = btcEvent.Args + if tc.Arguments == nil { + tc.Arguments = make(map[string]any) + } + + // Keep persisted assistant tool-call arguments aligned with rewritten execution args. + updateToolCallArguments(&messages[assistantMsgIndex], tcIdx, tc.Arguments) + if assistantSessionIndex >= 0 { + history := agent.Sessions.GetHistory(opts.SessionKey) + if assistantSessionIndex < len(history) { + updateToolCallArguments(&history[assistantSessionIndex], tcIdx, tc.Arguments) + agent.Sessions.SetHistory(opts.SessionKey, history) + } + } + } + + var toolDuration time.Duration + if !toolCanceled { + toolStart := time.Now() + toolResult = agent.Tools.ExecuteWithContext( + ctx, + tc.Name, + tc.Arguments, + opts.Channel, + opts.ChatID, + asyncCallback, + ) + toolDuration = time.Since(toolStart) + } + + // Fire after_tool_call hook (fires for both executed and canceled calls) + if al.hooks != nil { + al.hooks.TriggerAfterToolCall(ctx, &hooks.AfterToolCallEvent{ + ToolName: tc.Name, + Args: tc.Arguments, + Channel: opts.Channel, + ChatID: opts.ChatID, + Duration: toolDuration, + Result: toolResult, + }) + } // Send ForUser content to user immediately if not Silent if !toolResult.Silent && toolResult.ForUser != "" && opts.SendResponse { - al.bus.PublishOutbound(ctx, bus.OutboundMessage{ + al.sendOutbound(ctx, bus.OutboundMessage{ Channel: opts.Channel, ChatID: opts.ChatID, Content: toolResult.ForUser, @@ -883,28 +915,6 @@ func (al *AgentLoop) runLLMIteration( }) } - // If tool returned media refs, publish them as outbound media - if len(toolResult.Media) > 0 && opts.SendResponse { - parts := make([]bus.MediaPart, 0, len(toolResult.Media)) - for _, ref := range toolResult.Media { - part := bus.MediaPart{Ref: ref} - // Populate metadata from MediaStore when available - if al.mediaStore != nil { - if _, meta, err := al.mediaStore.ResolveWithMeta(ref); err == nil { - part.Filename = meta.Filename - part.ContentType = meta.ContentType - part.Type = inferMediaType(meta.Filename, meta.ContentType) - } - } - parts = append(parts, part) - } - al.bus.PublishOutboundMedia(ctx, bus.OutboundMediaMessage{ - Channel: opts.Channel, - ChatID: opts.ChatID, - Parts: parts, - }) - } - // Determine content for LLM based on tool result contentForLLM := toolResult.ForLLM if contentForLLM == "" && toolResult.Err != nil { @@ -947,7 +957,7 @@ func (al *AgentLoop) updateToolContexts(agent *AgentInstance, channel, chatID st } // maybeSummarize triggers summarization if the session history exceeds thresholds. -func (al *AgentLoop) maybeSummarize(agent *AgentInstance, sessionKey, channel, chatID string) { +func (al *AgentLoop) maybeSummarize(_ context.Context, agent *AgentInstance, sessionKey, channel, chatID string) { newHistory := agent.Sessions.GetHistory(sessionKey) tokenEstimate := al.estimateTokens(newHistory) threshold := agent.ContextWindow * 75 / 100 @@ -957,6 +967,13 @@ func (al *AgentLoop) maybeSummarize(agent *AgentInstance, sessionKey, channel, c if _, loading := al.summarizing.LoadOrStore(summarizeKey, true); !loading { go func() { defer al.summarizing.Delete(summarizeKey) + if !constants.IsInternalChannel(channel) { + al.sendOutbound(context.Background(), bus.OutboundMessage{ + Channel: channel, + ChatID: chatID, + Content: "Memory threshold reached. Optimizing conversation history...", + }) + } logger.Debug("Memory threshold reached. Optimizing conversation history...") al.summarizeSession(agent, sessionKey) }() @@ -1026,6 +1043,14 @@ func (al *AgentLoop) GetStartupInfo() map[string]any { return info } + pluginNames := make([]string, 0) + if al.pluginManager != nil { + pluginNames = al.pluginManager.Names() + if pluginNames == nil { + pluginNames = make([]string, 0) + } + } + // Tools info toolsList := agent.Tools.List() info["tools"] = map[string]any{ @@ -1033,6 +1058,12 @@ func (al *AgentLoop) GetStartupInfo() map[string]any { "names": toolsList, } + // Plugins info + info["plugins"] = map[string]any{ + "enabled": pluginNames, + "count": len(pluginNames), + } + // Skills info info["skills"] = agent.ContextBuilder.GetSkillsInfo() @@ -1045,6 +1076,19 @@ func (al *AgentLoop) GetStartupInfo() map[string]any { return info } +// updateToolCallArguments patches the serialized arguments for a tool call in-place. +func updateToolCallArguments(msg *providers.Message, toolCallIndex int, args map[string]any) { + if msg == nil || toolCallIndex < 0 || toolCallIndex >= len(msg.ToolCalls) { + return + } + toolCall := &msg.ToolCalls[toolCallIndex] + if toolCall.Function == nil { + return + } + argumentsJSON, _ := json.Marshal(args) + toolCall.Function.Arguments = string(argumentsJSON) +} + // formatMessagesForLog formats messages for logging func formatMessagesForLog(messages []providers.Message) string { if len(messages) == 0 { @@ -1317,20 +1361,21 @@ func (al *AgentLoop) handleCommand(ctx context.Context, msg bus.InboundMessage) return "", false } -// extractPeer extracts the routing peer from the inbound message's structured Peer field. +// extractPeer extracts the routing peer from inbound message metadata. func extractPeer(msg bus.InboundMessage) *routing.RoutePeer { - if msg.Peer.Kind == "" { + peerKind := msg.Metadata["peer_kind"] + if peerKind == "" { return nil } - peerID := msg.Peer.ID + peerID := msg.Metadata["peer_id"] if peerID == "" { - if msg.Peer.Kind == "direct" { + if peerKind == "direct" { peerID = msg.SenderID } else { peerID = msg.ChatID } } - return &routing.RoutePeer{Kind: msg.Peer.Kind, ID: peerID} + return &routing.RoutePeer{Kind: peerKind, ID: peerID} } // extractParentPeer extracts the parent peer (reply-to) from inbound message metadata. diff --git a/pkg/agent/loop_test.go b/pkg/agent/loop_test.go index 801b6a46e..fc2beb390 100644 --- a/pkg/agent/loop_test.go +++ b/pkg/agent/loop_test.go @@ -321,6 +321,90 @@ func TestAgentLoop_GetStartupInfo(t *testing.T) { } } +func TestGetStartupInfo_IncludesPluginSummary(t *testing.T) { + newLoop := func(t *testing.T) *AgentLoop { + t.Helper() + tmpDir := t.TempDir() + cfg := &config.Config{ + Agents: config.AgentsConfig{ + Defaults: config.AgentDefaults{ + Workspace: tmpDir, + Model: "test-model", + MaxTokens: 4096, + MaxToolIterations: 10, + }, + }, + } + return NewAgentLoop(cfg, bus.NewMessageBus(), &mockProvider{}) + } + + t.Run("no plugins enabled", func(t *testing.T) { + al := newLoop(t) + info := al.GetStartupInfo() + + pluginsInfo, ok := info["plugins"].(map[string]any) + if !ok { + t.Fatal("Expected 'plugins' to be a map") + } + + count, ok := pluginsInfo["count"].(int) + if !ok { + t.Fatal("Expected plugin count to be an int") + } + if count != 0 { + t.Fatalf("Expected plugin count 0, got %d", count) + } + + enabled, ok := pluginsInfo["enabled"].([]string) + if !ok { + t.Fatal("Expected plugin enabled list to be []string") + } + if len(enabled) != 0 { + t.Fatalf("Expected no enabled plugins, got %v", enabled) + } + }) + + t.Run("plugins enabled", func(t *testing.T) { + al := newLoop(t) + if err := al.EnablePlugins(blockingPlugin{}); err != nil { + t.Fatalf("EnablePlugins failed: %v", err) + } + + info := al.GetStartupInfo() + pluginsInfo, ok := info["plugins"].(map[string]any) + if !ok { + t.Fatal("Expected 'plugins' to be a map") + } + + count, ok := pluginsInfo["count"].(int) + if !ok { + t.Fatal("Expected plugin count to be an int") + } + if count <= 0 { + t.Fatalf("Expected plugin count > 0, got %d", count) + } + + enabled, ok := pluginsInfo["enabled"].([]string) + if !ok { + t.Fatal("Expected plugin enabled list to be []string") + } + if len(enabled) == 0 { + t.Fatal("Expected at least one enabled plugin") + } + + found := false + for _, name := range enabled { + if name == "block-outbound" { + found = true + break + } + } + if !found { + t.Fatalf("Expected enabled plugin list to include block-outbound, got %v", enabled) + } + }) +} + // TestAgentLoop_Stop verifies Stop() sets running to false func TestAgentLoop_Stop(t *testing.T) { tmpDir, err := os.MkdirTemp("", "agent-test-*") diff --git a/pkg/agent/plugin_test.go b/pkg/agent/plugin_test.go new file mode 100644 index 000000000..c0b6f0625 --- /dev/null +++ b/pkg/agent/plugin_test.go @@ -0,0 +1,416 @@ +package agent + +import ( + "context" + "encoding/json" + "os" + "strings" + "testing" + "time" + + "github.com/sipeed/picoclaw/pkg/bus" + "github.com/sipeed/picoclaw/pkg/config" + "github.com/sipeed/picoclaw/pkg/hooks" + "github.com/sipeed/picoclaw/pkg/plugin" + "github.com/sipeed/picoclaw/pkg/providers" + "github.com/sipeed/picoclaw/pkg/tools" +) + +type blockingPlugin struct{} + +func (p blockingPlugin) Name() string { + return "block-outbound" +} + +func (p blockingPlugin) APIVersion() string { + return plugin.APIVersion +} + +func (p blockingPlugin) Register(r *hooks.HookRegistry) error { + r.OnMessageSending("block-all", 0, func(_ context.Context, e *hooks.MessageSendingEvent) error { + e.Cancel = true + e.CancelReason = "blocked by plugin" + return nil + }) + return nil +} + +type nilArgsProvider struct { + calls int +} + +func (p *nilArgsProvider) Chat( + _ context.Context, + _ []providers.Message, + _ []providers.ToolDefinition, + _ string, + _ map[string]any, +) (*providers.LLMResponse, error) { + if p.calls == 0 { + p.calls++ + return &providers.LLMResponse{ + Content: "", + ToolCalls: []providers.ToolCall{ + { + ID: "tc-1", + Type: "function", + Name: "nil_args_tool", + Arguments: map[string]any{"seed": "value"}, + }, + }, + }, nil + } + p.calls++ + return &providers.LLMResponse{ + Content: "done", + ToolCalls: []providers.ToolCall{}, + }, nil +} + +func (p *nilArgsProvider) GetDefaultModel() string { + return "test-model" +} + +type nilArgsCaptureTool struct { + receivedNil bool +} + +func (t *nilArgsCaptureTool) Name() string { + return "nil_args_tool" +} + +func (t *nilArgsCaptureTool) Description() string { + return "captures whether args are nil" +} + +func (t *nilArgsCaptureTool) Parameters() map[string]any { + return map[string]any{ + "type": "object", + "properties": map[string]any{}, + } +} + +func (t *nilArgsCaptureTool) Execute(_ context.Context, args map[string]any) *tools.ToolResult { + if args == nil { + t.receivedNil = true + } + return tools.SilentResult("ok") +} + +func TestSetPluginManagerInstallsHookRegistry(t *testing.T) { + tmpDir, err := os.MkdirTemp("", "agent-plugin-test-*") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tmpDir) + + cfg := &config.Config{ + Agents: config.AgentsConfig{ + Defaults: config.AgentDefaults{ + Workspace: tmpDir, + Model: "test-model", + MaxTokens: 4096, + MaxToolIterations: 10, + }, + }, + } + + msgBus := bus.NewMessageBus() + al := NewAgentLoop(cfg, msgBus, &mockProvider{}) + + pm := plugin.NewManager() + if err := pm.Register(blockingPlugin{}); err != nil { + t.Fatalf("register plugin: %v", err) + } + + if err := al.SetPluginManager(pm); err != nil { + t.Fatalf("SetPluginManager: %v", err) + } + + if al.pluginManager == nil { + t.Fatal("expected plugin manager to be set") + } + if al.hooks != pm.HookRegistry() { + t.Fatal("expected agent loop hooks to use plugin manager registry") + } + + sent, reason := al.sendOutbound(context.Background(), bus.OutboundMessage{ + Channel: "cli", + ChatID: "direct", + Content: "hello", + }) + if sent { + t.Fatal("expected outbound message to be blocked by plugin") + } + if reason == "" { + t.Fatal("expected cancel reason to be propagated") + } +} + +func TestSetHooksReturnsErrorWhenRunning(t *testing.T) { + tmpDir, err := os.MkdirTemp("", "agent-plugin-test-*") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tmpDir) + + cfg := &config.Config{ + Agents: config.AgentsConfig{ + Defaults: config.AgentDefaults{ + Workspace: tmpDir, + Model: "test-model", + MaxTokens: 4096, + MaxToolIterations: 10, + }, + }, + } + + msgBus := bus.NewMessageBus() + al := NewAgentLoop(cfg, msgBus, &mockProvider{}) + al.running.Store(true) + + if err := al.SetHooks(hooks.NewHookRegistry()); err == nil { + t.Fatal("expected error when calling SetHooks while running") + } +} + +func TestSetPluginManagerDoesNotPartiallyUpdateOnError(t *testing.T) { + tmpDir, err := os.MkdirTemp("", "agent-plugin-test-*") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tmpDir) + + cfg := &config.Config{ + Agents: config.AgentsConfig{ + Defaults: config.AgentDefaults{ + Workspace: tmpDir, + Model: "test-model", + MaxTokens: 4096, + MaxToolIterations: 10, + }, + }, + } + + msgBus := bus.NewMessageBus() + al := NewAgentLoop(cfg, msgBus, &mockProvider{}) + al.running.Store(true) + + pm := plugin.NewManager() + if err := pm.Register(blockingPlugin{}); err != nil { + t.Fatalf("register plugin: %v", err) + } + + if err := al.SetPluginManager(pm); err == nil { + t.Fatal("expected SetPluginManager to fail while running") + } + if al.pluginManager != nil { + t.Fatal("expected plugin manager to remain unchanged on SetPluginManager failure") + } +} + +func TestBeforeToolCallHooksCannotLeaveToolArgsNil(t *testing.T) { + tmpDir, err := os.MkdirTemp("", "agent-plugin-test-*") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tmpDir) + + cfg := &config.Config{ + Agents: config.AgentsConfig{ + Defaults: config.AgentDefaults{ + Workspace: tmpDir, + Model: "test-model", + MaxTokens: 4096, + MaxToolIterations: 10, + }, + }, + } + + msgBus := bus.NewMessageBus() + provider := &nilArgsProvider{} + al := NewAgentLoop(cfg, msgBus, provider) + + captureTool := &nilArgsCaptureTool{} + al.RegisterTool(captureTool) + + r := hooks.NewHookRegistry() + r.OnBeforeToolCall("force-nil-args", 0, func(_ context.Context, e *hooks.BeforeToolCallEvent) error { + if e.ToolName == "nil_args_tool" { + e.Args = nil + } + return nil + }) + if setErr := al.SetHooks(r); setErr != nil { + t.Fatalf("SetHooks: %v", setErr) + } + + resp, err := al.ProcessDirectWithChannel(context.Background(), "run nil args test", "s1", "cli", "direct") + if err != nil { + t.Fatalf("ProcessDirectWithChannel: %v", err) + } + if resp != "done" { + t.Fatalf("expected final response 'done', got %q", resp) + } + if captureTool.receivedNil { + t.Fatal("expected tool args to be reinitialized to non-nil map") + } +} + +func TestSetHooksNilRestoresDirectMessageCallback(t *testing.T) { + tmpDir, err := os.MkdirTemp("", "agent-plugin-test-*") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tmpDir) + + cfg := &config.Config{ + Agents: config.AgentsConfig{ + Defaults: config.AgentDefaults{ + Workspace: tmpDir, + Model: "test-model", + MaxTokens: 4096, + MaxToolIterations: 10, + }, + }, + } + + msgBus := bus.NewMessageBus() + al := NewAgentLoop(cfg, msgBus, &mockProvider{}) + agent := al.registry.GetDefaultAgent() + if agent == nil { + t.Fatal("expected default agent") + } + tool, ok := agent.Tools.Get("message") + if !ok { + t.Fatal("expected message tool") + } + mt, ok := tool.(*tools.MessageTool) + if !ok { + t.Fatal("expected message tool type") + } + + reg := hooks.NewHookRegistry() + reg.OnMessageSending("block-all", 0, func(_ context.Context, e *hooks.MessageSendingEvent) error { + e.Cancel = true + e.CancelReason = "blocked-by-hook" + return nil + }) + if err := al.SetHooks(reg); err != nil { + t.Fatalf("SetHooks(reg): %v", err) + } + + blocked := mt.Execute(context.Background(), map[string]any{ + "content": "first", + "channel": "cli", + "chat_id": "direct", + }) + if !blocked.IsError { + t.Fatal("expected message tool call to fail while hooks are active") + } + if blocked.Err == nil || !strings.Contains(blocked.Err.Error(), "blocked-by-hook") { + t.Fatalf("expected hook cancel reason in error, got %#v", blocked.Err) + } + + ctxNoMsg, cancelNoMsg := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancelNoMsg() + if _, got := msgBus.SubscribeOutbound(ctxNoMsg); got { + t.Fatal("did not expect outbound message while hook cancellation is active") + } + + if err := al.SetHooks(nil); err != nil { + t.Fatalf("SetHooks(nil): %v", err) + } + + delivered := mt.Execute(context.Background(), map[string]any{ + "content": "second", + "channel": "cli", + "chat_id": "direct", + }) + if delivered.IsError { + t.Fatalf("expected message tool to succeed after SetHooks(nil), got %#v", delivered) + } + + ctxMsg, cancelMsg := context.WithTimeout(context.Background(), time.Second) + defer cancelMsg() + msg, got := msgBus.SubscribeOutbound(ctxMsg) + if !got { + t.Fatal("expected outbound message after SetHooks(nil)") + } + if msg.Content != "second" || msg.Channel != "cli" || msg.ChatID != "direct" { + t.Fatalf("unexpected outbound message: %#v", msg) + } +} + +func TestBeforeToolCallArgRewriteUpdatesAssistantTranscript(t *testing.T) { + tmpDir, err := os.MkdirTemp("", "agent-plugin-test-*") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tmpDir) + + cfg := &config.Config{ + Agents: config.AgentsConfig{ + Defaults: config.AgentDefaults{ + Workspace: tmpDir, + Model: "test-model", + MaxTokens: 4096, + MaxToolIterations: 10, + }, + }, + } + + msgBus := bus.NewMessageBus() + provider := &nilArgsProvider{} + al := NewAgentLoop(cfg, msgBus, provider) + al.RegisterTool(&nilArgsCaptureTool{}) + defaultAgent := al.registry.GetDefaultAgent() + if defaultAgent == nil { + t.Fatal("expected default agent") + } + sessionKey := "agent:" + defaultAgent.ID + ":s2" + + reg := hooks.NewHookRegistry() + reg.OnBeforeToolCall("rewrite-args", 0, func(_ context.Context, e *hooks.BeforeToolCallEvent) error { + e.Args["rewritten"] = "yes" + return nil + }) + if err := al.SetHooks(reg); err != nil { + t.Fatalf("SetHooks: %v", err) + } + + if _, err := al.ProcessDirectWithChannel( + context.Background(), + "run rewrite test", + sessionKey, + "cli", + "direct", + ); err != nil { + t.Fatalf("ProcessDirectWithChannel: %v", err) + } + + history := defaultAgent.Sessions.GetHistory(sessionKey) + + foundToolCall := false + for _, msg := range history { + if msg.Role != "assistant" || len(msg.ToolCalls) == 0 { + continue + } + if msg.ToolCalls[0].Function == nil { + t.Fatal("expected tool call function payload") + } + var args map[string]any + if err := json.Unmarshal([]byte(msg.ToolCalls[0].Function.Arguments), &args); err != nil { + t.Fatalf("failed to decode persisted tool call args: %v", err) + } + if got := args["rewritten"]; got != "yes" { + t.Fatalf("expected rewritten arg to be persisted, got %#v", got) + } + foundToolCall = true + break + } + + if !foundToolCall { + t.Fatal("expected assistant tool call message in session history") + } +}