add stop command

This commit is contained in:
t 2026-03-09 15:59:34 +08:00
parent b914e85352
commit 771e53e857
7 changed files with 140 additions and 60 deletions

View file

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

View file

@ -12,5 +12,6 @@ func BuiltinDefinitions() []Definition {
listCommand(),
switchCommand(),
checkCommand(),
stopCommand(),
}
}

23
pkg/commands/cmd_stop.go Normal file
View file

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

View file

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

View file

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

View file

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

View file

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