diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index 5840bb2ce..6fe968d10 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -357,16 +357,32 @@ func (al *AgentLoop) Run(ctx context.Context) error { } } - for al.running.Load() { - select { - case <-ctx.Done(): - return nil - default: + // Use a channel to decouple message consumption from message processing. + // This allows the main loop to continuously consume messages (including /stop) + // while processing happens in separate goroutines. + msgChan := make(chan bus.InboundMessage, 16) + + // Start message consumer goroutine + go func() { + for al.running.Load() { msg, ok := al.bus.ConsumeInbound(ctx) if !ok { continue } + select { + case msgChan <- msg: + case <-ctx.Done(): + return + } + } + }() + // Main loop: process messages from channel + for al.running.Load() { + select { + case <-ctx.Done(): + return nil + case msg := <-msgChan: // Check if this is a command - commands are handled asynchronously // so they can interrupt long-running tasks (e.g., /stop) if commands.HasCommandPrefix(msg.Content) { @@ -374,61 +390,9 @@ func (al *AgentLoop) Run(ctx context.Context) error { continue } - // Process non-command message synchronously - func() { - // Create cancellable context for this message - msgCtx, msgCancel := context.WithCancel(ctx) - defer msgCancel() - - // Store cancel function for /stop command - al.setCurrentCancel(msgCancel) - defer al.clearCurrentCancel() - - response, err := al.processMessage(msgCtx, msg) - if err != nil { - // Check if the error is due to context cancellation (user issued /stop) - if errors.Is(err, context.Canceled) { - response = "⏹️ Task stopped." - } else { - 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}, - ) - } - } - }() + // Process non-command message in a goroutine + // This allows the main loop to continue consuming messages + go al.processMessageAsync(ctx, msg) } } @@ -1730,6 +1694,65 @@ func (al *AgentLoop) estimateTokens(messages []providers.Message) int { return totalChars * 2 / 5 } +// processMessageAsync processes a non-command message in a goroutine. +// This allows the main loop to continue consuming messages while processing. +func (al *AgentLoop) processMessageAsync(ctx context.Context, msg bus.InboundMessage) { + // Create cancellable context for this message + msgCtx, msgCancel := context.WithCancel(ctx) + defer msgCancel() + + // Store cancel function for /stop command + al.setCurrentCancel(msgCancel) + defer al.clearCurrentCancel() + + response, err := al.processMessage(msgCtx, msg) + if err != nil { + // Check if the error is due to context cancellation (user issued /stop) + if errors.Is(err, context.Canceled) { + response = "⏹️ Task stopped." + } else { + 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 { + pubCtx, pubCancel := context.WithTimeout(context.Background(), 5*time.Second) + defer pubCancel() + al.bus.PublishOutbound(pubCtx, 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}, + ) + } + } +} + // handleCommandAsync handles commands in a separate goroutine so they can // interrupt long-running tasks. Commands like /stop need to be processed // immediately without waiting for the current message to finish. diff --git a/pkg/commands/builtin.go b/pkg/commands/builtin.go index a36dd3eba..99e8e052c 100644 --- a/pkg/commands/builtin.go +++ b/pkg/commands/builtin.go @@ -12,5 +12,6 @@ func BuiltinDefinitions() []Definition { listCommand(), switchCommand(), checkCommand(), + stopCommand(), } } diff --git a/pkg/commands/cmd_stop.go b/pkg/commands/cmd_stop.go new file mode 100644 index 000000000..86736ec30 --- /dev/null +++ b/pkg/commands/cmd_stop.go @@ -0,0 +1,23 @@ +package commands + +import "context" + +func stopCommand() Definition { + return Definition{ + Name: "stop", + Description: "Stop the current running task", + Usage: "/stop", + Strict: true, + Handler: func(_ context.Context, req Request, rt *Runtime) error { + if rt.CancelCurrentTask == nil { + return req.Reply("Stop command is not available in this context.") + } + if rt.CancelCurrentTask() { + // Don't send response here - processMessageAsync will send "Task stopped" + // when it detects the context cancellation. + return nil + } + return req.Reply("No task is currently running.") + }, + } +} \ No newline at end of file diff --git a/pkg/commands/definition.go b/pkg/commands/definition.go index 7309df317..53f7ff323 100644 --- a/pkg/commands/definition.go +++ b/pkg/commands/definition.go @@ -27,6 +27,7 @@ type Definition struct { Aliases []string SubCommands []SubCommand // optional; when set, Executor routes to sub-command handlers Handler Handler // for simple commands without sub-commands + Strict bool // if true, command must match exactly (no extra arguments) } // EffectiveUsage returns the usage string. When SubCommands are present, diff --git a/pkg/commands/executor.go b/pkg/commands/executor.go index 78a50e6c2..5323a9999 100644 --- a/pkg/commands/executor.go +++ b/pkg/commands/executor.go @@ -3,6 +3,7 @@ package commands import ( "context" "fmt" + "strings" ) type Outcome int @@ -56,6 +57,15 @@ func (e *Executor) executeDefinition(ctx context.Context, req Request, def Defin req.Reply = func(string) error { return nil } } + // Check if strict command has extra arguments + if def.Strict { + tokens := strings.Fields(strings.TrimSpace(req.Text)) + if len(tokens) > 1 { + // Strict command with extra arguments - passthrough to agent + return ExecuteResult{Outcome: OutcomePassthrough, Command: def.Name} + } + } + // Simple command — no sub-commands if len(def.SubCommands) == 0 { if def.Handler == nil { diff --git a/pkg/commands/runtime.go b/pkg/commands/runtime.go index 227d495f4..f718029f5 100644 --- a/pkg/commands/runtime.go +++ b/pkg/commands/runtime.go @@ -13,4 +13,5 @@ type Runtime struct { GetEnabledChannels func() []string SwitchModel func(value string) (oldModel string, err error) SwitchChannel func(value string) error + CancelCurrentTask func() bool // Cancel the currently running task, returns true if a task was cancelled } diff --git a/pkg/config/config.go b/pkg/config/config.go index 1beffe3cc..b4ae70c27 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -663,6 +663,7 @@ type ToolsConfig struct { MCP MCPConfig `json:"mcp"` LLMCallLog LLMCallLogConfig `json:"llm_call_log"` ConversationLog ConversationLogConfig `json:"conversation_log"` + Sanitizer SanitizerConfig `json:"sanitizer"` AppendFile ToolConfig `json:"append_file" envPrefix:"PICOCLAW_TOOLS_APPEND_FILE_"` EditFile ToolConfig `json:"edit_file" envPrefix:"PICOCLAW_TOOLS_EDIT_FILE_"` FindSkills ToolConfig `json:"find_skills" envPrefix:"PICOCLAW_TOOLS_FIND_SKILLS_"` @@ -693,6 +694,26 @@ type ConversationLogConfig struct { MaxFiles int `json:"max_files" env:"PICOCLAW_TOOLS_CONVERSATION_LOG_MAX_FILES"` } +// SanitizerConfig 配置敏感信息脱敏 +type SanitizerConfig struct { + Enabled bool `json:"enabled" env:"PICOCLAW_TOOLS_SANITIZER_ENABLED"` + Keywords []SanitizerKeyword `json:"keywords"` + CustomPatterns []SanitizerPattern `json:"custom_patterns"` +} + +// SanitizerKeyword 关键词脱敏规则 +type SanitizerKeyword struct { + Word string `json:"word"` + Tag string `json:"tag"` +} + +// SanitizerPattern 自定义正则脱敏规则 +type SanitizerPattern struct { + Name string `json:"name"` + Pattern string `json:"pattern"` + Tag string `json:"tag"` +} + type SearchCacheConfig struct { MaxSize int `json:"max_size" env:"PICOCLAW_SKILLS_SEARCH_CACHE_MAX_SIZE"` TTLSeconds int `json:"ttl_seconds" env:"PICOCLAW_SKILLS_SEARCH_CACHE_TTL_SECONDS"`