From 396186e7127250046ff584147786831841bb63b5 Mon Sep 17 00:00:00 2001 From: dj-oyu <68707227+dj-oyu@users.noreply.github.com> Date: Sun, 15 Mar 2026 18:36:45 +0900 Subject: [PATCH] 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) --- pkg/agent/loop.go | 79 +++---------------------------------------- pkg/agent/loop_ext.go | 71 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 75 insertions(+), 75 deletions(-) diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index da3e5bc66..1a6e42e46 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -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 : 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 diff --git a/pkg/agent/loop_ext.go b/pkg/agent/loop_ext.go index 67d99a89b..0be7b7def 100644 --- a/pkg/agent/loop_ext.go +++ b/pkg/agent/loop_ext.go @@ -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 +}