refactor: extract processMessage fork insertions to loop_ext.go
Move task intervention handling to handleTaskIntervention() and skill/plan expansion to expandForkCommands(). processMessage now calls these ext methods with 2 lines instead of ~75 inline lines. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
dj-oyu
parent
0fe20ed447
commit
396186e712
2 changed files with 75 additions and 75 deletions
|
|
@ -1111,64 +1111,8 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage)
|
|||
}
|
||||
|
||||
// Handle reply-based intervention for active tasks
|
||||
|
||||
if taskID, ok := msg.Metadata["task_id"]; ok && taskID != "" {
|
||||
if val, found := al.activeTasks.Load(taskID); found {
|
||||
task := val.(*activeTask)
|
||||
|
||||
content := strings.TrimSpace(msg.Content)
|
||||
|
||||
lower := strings.ToLower(content)
|
||||
|
||||
// Check for stop keywords
|
||||
|
||||
stopKeywords := []string{
|
||||
"stop", "cancel", "abort",
|
||||
|
||||
"停止", "中止", "やめて", //nolint:gosmopolitan // intentional CJK stop words
|
||||
|
||||
}
|
||||
|
||||
isStop := false
|
||||
|
||||
for _, kw := range stopKeywords {
|
||||
if lower == kw {
|
||||
isStop = true
|
||||
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
if isStop {
|
||||
task.cancel()
|
||||
|
||||
logger.InfoCF("agent", "Task canceled by user intervention",
|
||||
|
||||
map[string]any{"task_id": taskID})
|
||||
|
||||
return "Task canceled.", nil
|
||||
}
|
||||
|
||||
// Inject message into interrupt channel for the tool loop
|
||||
|
||||
select {
|
||||
case task.interrupt <- content:
|
||||
|
||||
logger.InfoCF("agent", "User intervention queued",
|
||||
|
||||
map[string]any{"task_id": taskID, "content": utils.Truncate(content, 80)})
|
||||
|
||||
default:
|
||||
|
||||
logger.WarnCF("agent", "Interrupt channel full, message dropped",
|
||||
|
||||
map[string]any{"task_id": taskID})
|
||||
}
|
||||
|
||||
return "Intervention sent to running task.", nil
|
||||
}
|
||||
|
||||
// Task not found — fall through to normal processing
|
||||
if response, handled := al.handleTaskIntervention(msg); handled {
|
||||
return response, nil
|
||||
}
|
||||
|
||||
// Route system messages to processSystemMessage
|
||||
|
|
@ -1182,23 +1126,8 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage)
|
|||
al.OnUserMessage()
|
||||
}
|
||||
|
||||
// Expand /skill command: inject SKILL.md content into message, then continue to LLM
|
||||
|
||||
var expansionCompact string
|
||||
|
||||
if expanded, compact, ok := al.expandSkillCommand(msg); ok {
|
||||
msg.Content = expanded
|
||||
|
||||
expansionCompact = compact
|
||||
}
|
||||
|
||||
// Expand /plan <task>: write interview seed, rewrite for LLM interview
|
||||
|
||||
if expanded, compact, ok := al.expandPlanCommand(msg); ok {
|
||||
msg.Content = expanded
|
||||
|
||||
expansionCompact = compact
|
||||
}
|
||||
// Expand fork-specific /skill and /plan commands
|
||||
expansionCompact := al.expandForkCommands(&msg)
|
||||
|
||||
// Check for commands
|
||||
|
||||
|
|
|
|||
|
|
@ -1,11 +1,15 @@
|
|||
package agent
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"github.com/sipeed/picoclaw/pkg/bus"
|
||||
"github.com/sipeed/picoclaw/pkg/config"
|
||||
"github.com/sipeed/picoclaw/pkg/logger"
|
||||
"github.com/sipeed/picoclaw/pkg/orch"
|
||||
"github.com/sipeed/picoclaw/pkg/stats"
|
||||
"github.com/sipeed/picoclaw/pkg/utils"
|
||||
)
|
||||
|
||||
// loopExt holds fork-specific fields for AgentLoop.
|
||||
|
|
@ -44,3 +48,70 @@ func (al *AgentLoop) SetConfigSaver(fn func(*config.Config) error) {
|
|||
func (al *AgentLoop) SetHeartbeatThreadUpdater(fn func(int)) {
|
||||
al.onHeartbeatThreadUpdate = fn
|
||||
}
|
||||
|
||||
// handleTaskIntervention checks if a message is a reply to an active task and
|
||||
// either cancels the task or injects a user intervention. Returns (response, handled).
|
||||
func (al *AgentLoop) handleTaskIntervention(msg bus.InboundMessage) (string, bool) {
|
||||
taskID, ok := msg.Metadata["task_id"]
|
||||
if !ok || taskID == "" {
|
||||
return "", false
|
||||
}
|
||||
|
||||
val, found := al.activeTasks.Load(taskID)
|
||||
if !found {
|
||||
// Task not found — fall through to normal processing
|
||||
return "", false
|
||||
}
|
||||
|
||||
task := val.(*activeTask)
|
||||
|
||||
content := strings.TrimSpace(msg.Content)
|
||||
lower := strings.ToLower(content)
|
||||
|
||||
// Check for stop keywords
|
||||
stopKeywords := []string{
|
||||
"stop", "cancel", "abort",
|
||||
"停止", "中止", "やめて", //nolint:gosmopolitan // intentional CJK stop words
|
||||
}
|
||||
|
||||
for _, kw := range stopKeywords {
|
||||
if lower == kw {
|
||||
task.cancel()
|
||||
|
||||
logger.InfoCF("agent", "Task canceled by user intervention",
|
||||
map[string]any{"task_id": taskID})
|
||||
|
||||
return "Task canceled.", true
|
||||
}
|
||||
}
|
||||
|
||||
// Inject message into interrupt channel for the tool loop
|
||||
select {
|
||||
case task.interrupt <- content:
|
||||
logger.InfoCF("agent", "User intervention queued",
|
||||
map[string]any{"task_id": taskID, "content": utils.Truncate(content, 80)})
|
||||
default:
|
||||
logger.WarnCF("agent", "Interrupt channel full, message dropped",
|
||||
map[string]any{"task_id": taskID})
|
||||
}
|
||||
|
||||
return "Intervention sent to running task.", true
|
||||
}
|
||||
|
||||
// expandForkCommands expands fork-specific /skill and /plan commands in the message.
|
||||
// Returns the modified message and the compact form for history.
|
||||
func (al *AgentLoop) expandForkCommands(msg *bus.InboundMessage) string {
|
||||
var expansionCompact string
|
||||
|
||||
if expanded, compact, ok := al.expandSkillCommand(*msg); ok {
|
||||
msg.Content = expanded
|
||||
expansionCompact = compact
|
||||
}
|
||||
|
||||
if expanded, compact, ok := al.expandPlanCommand(*msg); ok {
|
||||
msg.Content = expanded
|
||||
expansionCompact = compact
|
||||
}
|
||||
|
||||
return expansionCompact
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue