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

feat(agent): wire plugin bootstrap into agent and gateway

Browse source
This commit is contained in:
xj 2026-02-28 17:38:53 -08:00
parent bc6d6b1200
commit 3cceb99d22
5 changed files with 821 additions and 239 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

19
cmd/picoclaw/internal/agent/helpers.go
View file

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

18
cmd/picoclaw/internal/gateway/helpers.go
View file

@ -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:")

465
pkg/agent/loop.go
View file

@ -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,21 +169,6 @@ 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)
@ -207,26 +189,13 @@ func (al *AgentLoop) Run(ctx context.Context) error {
}
if !alreadySent {
al.bus.PublishOutbound(ctx, bus.OutboundMessage{
al.sendOutbound(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},
)
}
}
}()
}
}
@ -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"
}
return "file"
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
}
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,7 +836,49 @@ func (al *AgentLoop) runLLMIteration(
}
}
toolResult := agent.Tools.ExecuteWithContext(
// 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,
@ -868,10 +886,24 @@ func (al *AgentLoop) runLLMIteration(
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.

84
pkg/agent/loop_test.go
View file

@ -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-*")

416
pkg/agent/plugin_test.go Normal file
View file

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