refactor: split loop.go and tools into upstream-friendly file layout
Reduce upstream merge friction by extracting fork-specific code from files that upstream also modifies. loop.go is reduced from 6522 to 3165 lines by moving self-contained functions to dedicated files: - loop_task.go: activeTask, status display, tool log formatting - loop_plan.go: plan mode, interview filtering, worktree commands - loop_orch.go: orchestration reporting, system messages, peer extraction - loop_commands.go: slash commands, session commands, skill commands - loop_streaming.go: streaming display, repetition detection, reasoning - loop_session.go: session locks, GC, summarization - loop_info.go: getters (startup info, plan info, context info) Similarly for tools/: - base.go → base.go (Tool interface only) + base_ext.go (AsyncExecutor, StatusProvider, context helpers) - registry.go → registry.go (core CRUD) + registry_ext.go (ExecuteWithContext, ToProviderDefs, GetSummaries) config/: - SubagentsConfig, PeerMatch moved to config_ext.go - Fix missing "strings" import (pre-existing bug) No behavior changes. All code stays in the same Go packages. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
dj-oyu
parent
7bfd5cc658
commit
77c4f69d2e
14 changed files with 3786 additions and 3706 deletions
3357
pkg/agent/loop.go
3357
pkg/agent/loop.go
File diff suppressed because it is too large
Load diff
717
pkg/agent/loop_commands.go
Normal file
717
pkg/agent/loop_commands.go
Normal file
|
|
@ -0,0 +1,717 @@
|
|||
package agent
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/sipeed/picoclaw/pkg/bus"
|
||||
"github.com/sipeed/picoclaw/pkg/session"
|
||||
"github.com/sipeed/picoclaw/pkg/stats"
|
||||
)
|
||||
|
||||
func (al *AgentLoop) handleCommand(ctx context.Context, msg bus.InboundMessage) (string, bool) {
|
||||
content := strings.TrimSpace(msg.Content)
|
||||
|
||||
if !strings.HasPrefix(content, "/") {
|
||||
return "", false
|
||||
}
|
||||
|
||||
parts := strings.Fields(content)
|
||||
|
||||
if len(parts) == 0 {
|
||||
return "", false
|
||||
}
|
||||
|
||||
cmd := parts[0]
|
||||
|
||||
args := parts[1:]
|
||||
|
||||
switch cmd {
|
||||
case "/show":
|
||||
|
||||
if len(args) < 1 {
|
||||
return "Usage: /show [model|channel|agents]", true
|
||||
}
|
||||
|
||||
switch args[0] {
|
||||
case "model":
|
||||
|
||||
defaultAgent := al.registry.GetDefaultAgent()
|
||||
|
||||
if defaultAgent == nil {
|
||||
return "No default agent configured", true
|
||||
}
|
||||
|
||||
return fmt.Sprintf("Current model: %s", defaultAgent.Model), true
|
||||
|
||||
case "channel":
|
||||
|
||||
return fmt.Sprintf("Current channel: %s", msg.Channel), true
|
||||
|
||||
case "agents":
|
||||
|
||||
agentIDs := al.registry.ListAgentIDs()
|
||||
|
||||
return fmt.Sprintf("Registered agents: %s", strings.Join(agentIDs, ", ")), true
|
||||
|
||||
default:
|
||||
|
||||
return fmt.Sprintf("Unknown show target: %s", args[0]), true
|
||||
}
|
||||
|
||||
case "/list":
|
||||
|
||||
if len(args) < 1 {
|
||||
return "Usage: /list [models|channels|agents]", true
|
||||
}
|
||||
|
||||
switch args[0] {
|
||||
case "models":
|
||||
|
||||
return "Available models: configured in config.json per agent", true
|
||||
|
||||
case "channels":
|
||||
|
||||
if al.channelManager == nil {
|
||||
return "Channel manager not initialized", true
|
||||
}
|
||||
|
||||
channels := al.channelManager.GetEnabledChannels()
|
||||
|
||||
if len(channels) == 0 {
|
||||
return "No channels enabled", true
|
||||
}
|
||||
|
||||
return fmt.Sprintf("Enabled channels: %s", strings.Join(channels, ", ")), true
|
||||
|
||||
case "agents":
|
||||
|
||||
agentIDs := al.registry.ListAgentIDs()
|
||||
|
||||
return fmt.Sprintf("Registered agents: %s", strings.Join(agentIDs, ", ")), true
|
||||
|
||||
default:
|
||||
|
||||
return fmt.Sprintf("Unknown list target: %s", args[0]), true
|
||||
}
|
||||
|
||||
case "/switch":
|
||||
|
||||
if len(args) < 3 || args[1] != "to" {
|
||||
return "Usage: /switch [model|channel] to <name>", true
|
||||
}
|
||||
|
||||
target := args[0]
|
||||
|
||||
value := args[2]
|
||||
|
||||
switch target {
|
||||
case "model":
|
||||
|
||||
defaultAgent := al.registry.GetDefaultAgent()
|
||||
|
||||
if defaultAgent == nil {
|
||||
return "No default agent configured", true
|
||||
}
|
||||
|
||||
oldModel := defaultAgent.Model
|
||||
|
||||
defaultAgent.Model = value
|
||||
|
||||
return fmt.Sprintf("Switched model from %s to %s", oldModel, value), true
|
||||
|
||||
case "channel":
|
||||
|
||||
if al.channelManager == nil {
|
||||
return "Channel manager not initialized", true
|
||||
}
|
||||
|
||||
if _, exists := al.channelManager.GetChannel(value); !exists && value != "cli" {
|
||||
return fmt.Sprintf("Channel '%s' not found or not enabled", value), true
|
||||
}
|
||||
|
||||
return fmt.Sprintf("Switched target channel to %s", value), true
|
||||
|
||||
default:
|
||||
|
||||
return fmt.Sprintf("Unknown switch target: %s", target), true
|
||||
}
|
||||
|
||||
case "/session":
|
||||
|
||||
return al.handleSessionCommand(args, msg.SessionKey), true
|
||||
|
||||
case "/skills":
|
||||
|
||||
return al.handleSkillsCommand(), true
|
||||
|
||||
case "/plan":
|
||||
|
||||
resp, handled := al.handlePlanCommand(args, msg.SessionKey)
|
||||
|
||||
if handled {
|
||||
al.notifyStateChange()
|
||||
}
|
||||
|
||||
return resp, handled
|
||||
|
||||
case "/heartbeat":
|
||||
|
||||
resp, handled := al.handleHeartbeatCommand(args, msg)
|
||||
|
||||
if handled {
|
||||
al.notifyStateChange()
|
||||
}
|
||||
|
||||
return resp, handled
|
||||
}
|
||||
|
||||
return "", false
|
||||
}
|
||||
|
||||
func (al *AgentLoop) handleHeartbeatCommand(args []string, msg bus.InboundMessage) (string, bool) {
|
||||
if len(args) == 0 {
|
||||
return "Usage: /heartbeat thread [here|off|<thread_id>]", true
|
||||
}
|
||||
|
||||
if args[0] != "thread" {
|
||||
return "Usage: /heartbeat thread [here|off|<thread_id>]", true
|
||||
}
|
||||
|
||||
if len(args) < 2 {
|
||||
return "Usage: /heartbeat thread [here|off|<thread_id>]", true
|
||||
}
|
||||
|
||||
if msg.Channel != "telegram" {
|
||||
return "/heartbeat thread is only supported from Telegram chats.", true
|
||||
}
|
||||
|
||||
baseChatID, currentThreadID := splitChatAndThread(msg.ChatID)
|
||||
|
||||
if baseChatID == "" {
|
||||
return "Unable to detect Telegram chat ID for heartbeat routing.", true
|
||||
}
|
||||
|
||||
arg := strings.ToLower(strings.TrimSpace(args[1]))
|
||||
|
||||
var threadID int
|
||||
|
||||
var err error
|
||||
|
||||
switch arg {
|
||||
case "off", "disable", "clear":
|
||||
|
||||
threadID = 0
|
||||
|
||||
case "here", "this":
|
||||
|
||||
if currentThreadID <= 0 {
|
||||
return "Current Telegram message is not in a thread. Usage: /heartbeat thread <thread_id>", true
|
||||
}
|
||||
|
||||
threadID = currentThreadID
|
||||
|
||||
default:
|
||||
|
||||
threadID, err = strconv.Atoi(arg)
|
||||
|
||||
if err != nil || threadID < 0 {
|
||||
return "Usage: /heartbeat thread [here|off|<thread_id>]", true
|
||||
}
|
||||
}
|
||||
|
||||
al.cfg.Channels.Telegram.HeartbeatThreadID = threadID
|
||||
|
||||
if al.state != nil {
|
||||
_ = al.state.SetHeartbeatTarget(fmt.Sprintf("telegram:%s", baseChatID))
|
||||
}
|
||||
|
||||
if al.onHeartbeatThreadUpdate != nil {
|
||||
al.onHeartbeatThreadUpdate(threadID)
|
||||
}
|
||||
|
||||
if al.saveConfig != nil {
|
||||
if err := al.saveConfig(al.cfg); err != nil {
|
||||
return fmt.Sprintf("Failed to persist config.json: %v", err), true
|
||||
}
|
||||
}
|
||||
|
||||
if threadID == 0 {
|
||||
return fmt.Sprintf("Heartbeat thread routing disabled for chat %s and saved to config.json.", baseChatID), true
|
||||
}
|
||||
|
||||
return fmt.Sprintf("Heartbeat thread set to %d for chat %s and saved to config.json.", threadID, baseChatID), true
|
||||
}
|
||||
|
||||
func splitChatAndThread(chatID string) (baseChatID string, threadID int) {
|
||||
baseChatID = strings.TrimSpace(chatID)
|
||||
|
||||
if baseChatID == "" {
|
||||
return "", 0
|
||||
}
|
||||
|
||||
if slash := strings.Index(baseChatID, "/"); slash >= 0 {
|
||||
threadPart := strings.TrimSpace(baseChatID[slash+1:])
|
||||
|
||||
baseChatID = strings.TrimSpace(baseChatID[:slash])
|
||||
|
||||
if tid, err := strconv.Atoi(threadPart); err == nil && tid > 0 {
|
||||
threadID = tid
|
||||
}
|
||||
}
|
||||
|
||||
return baseChatID, threadID
|
||||
}
|
||||
|
||||
// handleSessionCommand dispatches /session subcommands.
|
||||
|
||||
func (al *AgentLoop) handleSessionCommand(args []string, sessionKey string) string {
|
||||
sub := ""
|
||||
|
||||
if len(args) > 0 {
|
||||
sub = strings.ToLower(strings.TrimSpace(args[0]))
|
||||
}
|
||||
|
||||
switch sub {
|
||||
case "list":
|
||||
|
||||
return al.handleSessionList()
|
||||
|
||||
case "graph":
|
||||
|
||||
return al.handleSessionGraph()
|
||||
|
||||
case "fork":
|
||||
|
||||
return al.handleSessionFork(args[1:], sessionKey)
|
||||
|
||||
case "reset":
|
||||
|
||||
if al.stats == nil {
|
||||
return "Stats tracking is disabled."
|
||||
}
|
||||
|
||||
al.stats.Reset()
|
||||
|
||||
return "Session statistics have been reset."
|
||||
|
||||
default:
|
||||
|
||||
return al.handleSessionStats()
|
||||
}
|
||||
}
|
||||
|
||||
func (al *AgentLoop) handleSessionStats() string {
|
||||
agent := al.registry.GetDefaultAgent()
|
||||
|
||||
store := agent.Sessions.Store()
|
||||
|
||||
// Session DAG summary
|
||||
|
||||
sessions, _ := store.List(nil)
|
||||
|
||||
var sb strings.Builder
|
||||
|
||||
fmt.Fprintf(&sb, "Sessions: %d in store\n", len(sessions))
|
||||
|
||||
if len(sessions) > 0 {
|
||||
active, completed := 0, 0
|
||||
|
||||
for _, s := range sessions {
|
||||
switch s.Status {
|
||||
case "active":
|
||||
|
||||
active++
|
||||
|
||||
case "completed":
|
||||
|
||||
completed++
|
||||
}
|
||||
}
|
||||
|
||||
fmt.Fprintf(&sb, " active=%d completed=%d\n", active, completed)
|
||||
}
|
||||
|
||||
sb.WriteString("\nUse: /session list | graph | fork [label]\n")
|
||||
|
||||
// Token stats if available
|
||||
|
||||
if al.stats != nil {
|
||||
s := al.stats.GetStats()
|
||||
|
||||
fmt.Fprintf(&sb,
|
||||
|
||||
"\nToken Stats — Today (%s):\n Prompts: %d LLM calls: %d Tokens: %s (in: %s, out: %s)\n"+
|
||||
|
||||
"All time (since %s):\n Prompts: %d LLM calls: %d Tokens: %s (in: %s, out: %s)",
|
||||
|
||||
s.Today.Date,
|
||||
|
||||
s.Today.Prompts,
|
||||
|
||||
s.Today.Requests,
|
||||
|
||||
stats.FormatTokenCount(s.Today.TotalTokens),
|
||||
|
||||
stats.FormatTokenCount(s.Today.PromptTokens),
|
||||
|
||||
stats.FormatTokenCount(s.Today.CompletionTokens),
|
||||
|
||||
s.Since.Format("2006-01-02"),
|
||||
|
||||
s.TotalPrompts,
|
||||
|
||||
s.TotalRequests,
|
||||
|
||||
stats.FormatTokenCount(s.TotalTokens),
|
||||
|
||||
stats.FormatTokenCount(s.TotalPromptTokens),
|
||||
|
||||
stats.FormatTokenCount(s.TotalCompletionTokens),
|
||||
)
|
||||
}
|
||||
|
||||
return sb.String()
|
||||
}
|
||||
|
||||
// shortSessionKey truncates long session keys for display.
|
||||
|
||||
func shortSessionKey(key string) string {
|
||||
parts := strings.Split(key, ":")
|
||||
|
||||
if len(parts) > 2 {
|
||||
return strings.Join(parts[2:], ":")
|
||||
}
|
||||
|
||||
return key
|
||||
}
|
||||
|
||||
func (al *AgentLoop) handleSessionList() string {
|
||||
agent := al.registry.GetDefaultAgent()
|
||||
|
||||
store := agent.Sessions.Store()
|
||||
|
||||
sessions, err := store.List(nil)
|
||||
if err != nil {
|
||||
return fmt.Sprintf("Error listing sessions: %v", err)
|
||||
}
|
||||
|
||||
if len(sessions) == 0 {
|
||||
return "No sessions in store."
|
||||
}
|
||||
|
||||
var sb strings.Builder
|
||||
|
||||
fmt.Fprintf(&sb, "Sessions (%d)\n", len(sessions))
|
||||
|
||||
for _, s := range sessions {
|
||||
age := time.Since(s.UpdatedAt).Truncate(time.Second)
|
||||
|
||||
label := s.Label
|
||||
|
||||
if label == "" {
|
||||
label = shortSessionKey(s.Key)
|
||||
}
|
||||
|
||||
parent := ""
|
||||
|
||||
if s.ParentKey != "" {
|
||||
parent = " parent=" + shortSessionKey(s.ParentKey)
|
||||
}
|
||||
|
||||
fmt.Fprintf(&sb, "- %s [%s] (%s) turns=%d%s\n",
|
||||
|
||||
label, s.Status, age, s.TurnCount, parent)
|
||||
}
|
||||
|
||||
return sb.String()
|
||||
}
|
||||
|
||||
func (al *AgentLoop) handleSessionGraph() string {
|
||||
agent := al.registry.GetDefaultAgent()
|
||||
|
||||
store := agent.Sessions.Store()
|
||||
|
||||
sessions, err := store.List(nil)
|
||||
if err != nil {
|
||||
return fmt.Sprintf("Error listing sessions: %v", err)
|
||||
}
|
||||
|
||||
if len(sessions) == 0 {
|
||||
return "No sessions in store."
|
||||
}
|
||||
|
||||
// Build parent→children map and find roots
|
||||
|
||||
byKey := make(map[string]*session.SessionInfo, len(sessions))
|
||||
|
||||
children := make(map[string][]string)
|
||||
|
||||
var roots []string
|
||||
|
||||
for _, s := range sessions {
|
||||
byKey[s.Key] = s
|
||||
|
||||
if s.ParentKey == "" {
|
||||
roots = append(roots, s.Key)
|
||||
} else {
|
||||
children[s.ParentKey] = append(children[s.ParentKey], s.Key)
|
||||
}
|
||||
}
|
||||
|
||||
var sb strings.Builder
|
||||
|
||||
sb.WriteString("Session Graph\n")
|
||||
|
||||
for i, root := range roots {
|
||||
last := i == len(roots)-1
|
||||
|
||||
printSessionTree(&sb, root, byKey, children, "", last)
|
||||
}
|
||||
|
||||
return sb.String()
|
||||
}
|
||||
|
||||
func printSessionTree(
|
||||
sb *strings.Builder,
|
||||
key string,
|
||||
byKey map[string]*session.SessionInfo,
|
||||
children map[string][]string,
|
||||
prefix string,
|
||||
last bool,
|
||||
) {
|
||||
s := byKey[key]
|
||||
|
||||
if s == nil {
|
||||
return
|
||||
}
|
||||
|
||||
connector := "├── "
|
||||
|
||||
if last {
|
||||
connector = "└── "
|
||||
}
|
||||
|
||||
icon := "●"
|
||||
|
||||
if s.Status == "completed" {
|
||||
icon = "✓"
|
||||
}
|
||||
|
||||
label := s.Label
|
||||
|
||||
if label == "" {
|
||||
label = shortSessionKey(s.Key)
|
||||
}
|
||||
|
||||
fmt.Fprintf(sb, "%s%s%s %s (turns=%d)\n", prefix, connector, icon, label, s.TurnCount)
|
||||
|
||||
childPrefix := prefix + "│ "
|
||||
|
||||
if last {
|
||||
childPrefix = prefix + " "
|
||||
}
|
||||
|
||||
kids := children[key]
|
||||
|
||||
for i, childKey := range kids {
|
||||
printSessionTree(sb, childKey, byKey, children, childPrefix, i == len(kids)-1)
|
||||
}
|
||||
}
|
||||
|
||||
func (al *AgentLoop) handleSessionFork(args []string, sessionKey string) string {
|
||||
if sessionKey == "" {
|
||||
return "Cannot fork: no active session key."
|
||||
}
|
||||
|
||||
agent := al.registry.GetDefaultAgent()
|
||||
|
||||
store := agent.Sessions.Store()
|
||||
|
||||
label := "fork"
|
||||
|
||||
if len(args) > 0 {
|
||||
label = strings.Join(args, " ")
|
||||
}
|
||||
|
||||
childKey := sessionKey + ":fork:" + time.Now().Format("20060102T150405")
|
||||
|
||||
err := store.Fork(sessionKey, childKey, &session.CreateOpts{Label: label})
|
||||
if err != nil {
|
||||
return fmt.Sprintf("Fork failed: %v", err)
|
||||
}
|
||||
|
||||
return fmt.Sprintf(
|
||||
"Forked session\n parent: %s\n child: %s",
|
||||
shortSessionKey(sessionKey),
|
||||
shortSessionKey(childKey),
|
||||
)
|
||||
}
|
||||
|
||||
type SessionGraphNode struct {
|
||||
Key string `json:"key"`
|
||||
|
||||
Label string `json:"label"`
|
||||
|
||||
Status string `json:"status"`
|
||||
|
||||
Summary string `json:"summary"`
|
||||
|
||||
ParentKey string `json:"parent_key"`
|
||||
|
||||
ForkTurnID string `json:"fork_turn_id"`
|
||||
|
||||
TurnCount int `json:"turn_count"`
|
||||
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
|
||||
UpdatedAt time.Time `json:"updated_at"`
|
||||
}
|
||||
|
||||
// GetSessionGraph returns all sessions as a flat list of graph nodes.
|
||||
|
||||
func (al *AgentLoop) GetSessionGraph() []SessionGraphNode {
|
||||
agent := al.registry.GetDefaultAgent()
|
||||
|
||||
store := agent.Sessions.Store()
|
||||
|
||||
sessions, err := store.List(nil)
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
nodes := make([]SessionGraphNode, 0, len(sessions))
|
||||
|
||||
for _, s := range sessions {
|
||||
nodes = append(nodes, SessionGraphNode{
|
||||
Key: s.Key,
|
||||
|
||||
Label: s.Label,
|
||||
|
||||
Status: s.Status,
|
||||
|
||||
Summary: s.Summary,
|
||||
|
||||
ParentKey: s.ParentKey,
|
||||
|
||||
ForkTurnID: s.ForkTurnID,
|
||||
|
||||
TurnCount: s.TurnCount,
|
||||
|
||||
CreatedAt: s.CreatedAt,
|
||||
|
||||
UpdatedAt: s.UpdatedAt,
|
||||
})
|
||||
}
|
||||
|
||||
return nodes
|
||||
}
|
||||
|
||||
// expandSkillCommand detects "/skill <name> [message]" and returns:
|
||||
|
||||
// - expanded: full content with SKILL.md injected (for LLM)
|
||||
|
||||
// - compact: skill name tag + user message only (for history)
|
||||
|
||||
// - ok: whether expansion happened
|
||||
|
||||
func (al *AgentLoop) expandSkillCommand(msg bus.InboundMessage) (expanded string, compact string, ok bool) {
|
||||
content := strings.TrimSpace(msg.Content)
|
||||
|
||||
if !strings.HasPrefix(content, "/skill ") {
|
||||
return "", "", false
|
||||
}
|
||||
|
||||
// Parse: /skill <name> [message]
|
||||
|
||||
rest := strings.TrimSpace(content[7:]) // len("/skill ") == 7
|
||||
|
||||
parts := strings.SplitN(rest, " ", 2)
|
||||
|
||||
if len(parts) == 0 || parts[0] == "" {
|
||||
return "", "", false
|
||||
}
|
||||
|
||||
skillName := parts[0]
|
||||
|
||||
userMessage := ""
|
||||
|
||||
if len(parts) > 1 {
|
||||
userMessage = strings.TrimSpace(parts[1])
|
||||
}
|
||||
|
||||
agent := al.registry.GetDefaultAgent()
|
||||
|
||||
if agent == nil {
|
||||
return "", "", false
|
||||
}
|
||||
|
||||
skillContent, found := agent.ContextBuilder.LoadSkill(skillName)
|
||||
|
||||
if !found {
|
||||
return "", "", false
|
||||
}
|
||||
|
||||
tag := fmt.Sprintf("[Skill: %s]", skillName)
|
||||
|
||||
// Build expanded message: skill instructions + user message (for LLM)
|
||||
|
||||
var sb strings.Builder
|
||||
|
||||
sb.WriteString(tag)
|
||||
|
||||
sb.WriteString("\n\n")
|
||||
|
||||
sb.WriteString(skillContent)
|
||||
|
||||
if userMessage != "" {
|
||||
sb.WriteString("\n\n---\n\n")
|
||||
|
||||
sb.WriteString(userMessage)
|
||||
}
|
||||
|
||||
// Build compact form: skill name tag + user message only (for history)
|
||||
|
||||
compactForm := tag
|
||||
|
||||
if userMessage != "" {
|
||||
compactForm = tag + "\n" + userMessage
|
||||
}
|
||||
|
||||
return sb.String(), compactForm, true
|
||||
}
|
||||
|
||||
// handleSkillsCommand lists all available skills.
|
||||
|
||||
func (al *AgentLoop) handleSkillsCommand() string {
|
||||
agent := al.registry.GetDefaultAgent()
|
||||
|
||||
if agent == nil {
|
||||
return "No agent configured."
|
||||
}
|
||||
|
||||
skillsList := agent.ContextBuilder.ListSkills()
|
||||
|
||||
if len(skillsList) == 0 {
|
||||
return "No skills available.\nAdd skills to your workspace/skills/ directory."
|
||||
}
|
||||
|
||||
var sb strings.Builder
|
||||
|
||||
sb.WriteString("Available Skills\n\n")
|
||||
|
||||
for _, s := range skillsList {
|
||||
fmt.Fprintf(&sb, "**%s** (%s)\n", s.Name, s.Source)
|
||||
|
||||
if s.Description != "" {
|
||||
fmt.Fprintf(&sb, "```\n%s\n```\n", s.Description)
|
||||
}
|
||||
}
|
||||
|
||||
sb.WriteString("\nUse: /skill <name> [message]")
|
||||
|
||||
return sb.String()
|
||||
}
|
||||
281
pkg/agent/loop_info.go
Normal file
281
pkg/agent/loop_info.go
Normal file
|
|
@ -0,0 +1,281 @@
|
|||
package agent
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
"github.com/sipeed/picoclaw/pkg/providers"
|
||||
"github.com/sipeed/picoclaw/pkg/skills"
|
||||
"github.com/sipeed/picoclaw/pkg/stats"
|
||||
"github.com/sipeed/picoclaw/pkg/tools"
|
||||
"github.com/sipeed/picoclaw/pkg/utils"
|
||||
)
|
||||
|
||||
func (al *AgentLoop) GetStartupInfo() map[string]any {
|
||||
info := make(map[string]any)
|
||||
|
||||
agent := al.registry.GetDefaultAgent()
|
||||
|
||||
if agent == nil {
|
||||
return info
|
||||
}
|
||||
|
||||
// Tools info
|
||||
|
||||
toolsList := agent.Tools.List()
|
||||
|
||||
toolsMap := map[string]any{
|
||||
"count": len(toolsList),
|
||||
|
||||
"names": toolsList,
|
||||
}
|
||||
|
||||
// Report web search provider if registered
|
||||
|
||||
if t, ok := agent.Tools.Get("web_search"); ok {
|
||||
if wst, ok := t.(*tools.WebSearchTool); ok {
|
||||
toolsMap["web_search_provider"] = wst.ProviderName()
|
||||
}
|
||||
}
|
||||
|
||||
info["tools"] = toolsMap
|
||||
|
||||
// Skills info
|
||||
|
||||
info["skills"] = agent.ContextBuilder.GetSkillsInfo()
|
||||
|
||||
// Agents info
|
||||
|
||||
info["agents"] = map[string]any{
|
||||
"count": len(al.registry.ListAgentIDs()),
|
||||
|
||||
"ids": al.registry.ListAgentIDs(),
|
||||
}
|
||||
|
||||
return info
|
||||
}
|
||||
|
||||
// ListSkills returns all available skills from the default agent.
|
||||
|
||||
func (al *AgentLoop) ListSkills() []skills.SkillInfo {
|
||||
agent := al.registry.GetDefaultAgent()
|
||||
|
||||
if agent == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
return agent.ContextBuilder.ListSkills()
|
||||
}
|
||||
|
||||
// GetPlanInfo returns plan state from the default agent's memory store.
|
||||
|
||||
func (al *AgentLoop) GetPlanInfo() (hasPlan bool, status string, currentPhase, totalPhases int, display string, memory string) {
|
||||
agent := al.registry.GetDefaultAgent()
|
||||
|
||||
if agent == nil {
|
||||
return false, "", 0, 0, "No agent available.", ""
|
||||
}
|
||||
|
||||
mem := agent.ContextBuilder.Memory()
|
||||
|
||||
if mem == nil {
|
||||
return false, "", 0, 0, "No memory store.", ""
|
||||
}
|
||||
|
||||
hasPlan = mem.HasActivePlan()
|
||||
|
||||
status = mem.GetPlanStatus()
|
||||
|
||||
currentPhase = mem.GetCurrentPhase()
|
||||
|
||||
totalPhases = mem.GetTotalPhases()
|
||||
|
||||
display = mem.FormatPlanDisplay()
|
||||
|
||||
memory = mem.ReadLongTerm()
|
||||
|
||||
return hasPlan, status, currentPhase, totalPhases, display, memory
|
||||
}
|
||||
|
||||
// GetPlanStatus returns the current plan status ("interviewing", "executing", "review", etc.) or "".
|
||||
|
||||
func (al *AgentLoop) GetPlanStatus() string {
|
||||
agent := al.registry.GetDefaultAgent()
|
||||
|
||||
if agent == nil {
|
||||
return ""
|
||||
}
|
||||
|
||||
return agent.ContextBuilder.GetPlanStatus()
|
||||
}
|
||||
|
||||
// GetPlanPhases returns structured phase/step data from the default agent's plan.
|
||||
|
||||
func (al *AgentLoop) GetPlanPhases() []PlanPhase {
|
||||
agent := al.registry.GetDefaultAgent()
|
||||
|
||||
if agent == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
mem := agent.ContextBuilder.Memory()
|
||||
|
||||
if mem == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
return mem.GetPlanPhases()
|
||||
}
|
||||
|
||||
// GetActiveSessions returns currently active sessions for the mini app API.
|
||||
|
||||
func (al *AgentLoop) GetActiveSessions() []SessionEntry {
|
||||
return al.sessions.ListActive()
|
||||
}
|
||||
|
||||
// GetSessionStats returns the current session statistics snapshot, or nil if stats tracking is disabled.
|
||||
|
||||
func (al *AgentLoop) GetSessionStats() *stats.Stats {
|
||||
if al.stats == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
s := al.stats.GetStats()
|
||||
|
||||
return &s
|
||||
}
|
||||
|
||||
// GetContextInfo returns the bootstrap file resolution and directory context for the default agent.
|
||||
|
||||
func (al *AgentLoop) GetContextInfo() (workDir, planWorkDir, workspace string, bootstrap []BootstrapFileInfo) {
|
||||
agent := al.registry.GetDefaultAgent()
|
||||
|
||||
if agent == nil {
|
||||
return "", "", "", nil
|
||||
}
|
||||
|
||||
workspace = agent.Workspace
|
||||
|
||||
planWorkDir = agent.ContextBuilder.GetPlanWorkDir()
|
||||
|
||||
// Use the most recent active session's touch_dir (tool-detected project directory)
|
||||
|
||||
if active := al.sessions.ListActive(); len(active) > 0 && active[0].TouchDir != "" {
|
||||
workDir = active[0].TouchDir
|
||||
} else {
|
||||
workDir = agent.ContextBuilder.workDir
|
||||
}
|
||||
|
||||
bootstrap = agent.ContextBuilder.ResolveBootstrapPaths()
|
||||
|
||||
return workDir, planWorkDir, workspace, bootstrap
|
||||
}
|
||||
|
||||
// GetSystemPrompt returns the system prompt last sent to the LLM.
|
||||
|
||||
// If the prompt is dirty (state changed since last capture), it rebuilds
|
||||
|
||||
// from current state. Falls back to building if no LLM call has occurred yet.
|
||||
|
||||
func (al *AgentLoop) GetSystemPrompt() string {
|
||||
if !al.promptDirty.Load() {
|
||||
if v := al.lastSystemPrompt.Load(); v != nil {
|
||||
return v.(string)
|
||||
}
|
||||
}
|
||||
|
||||
// Rebuild from current state
|
||||
|
||||
agent := al.registry.GetDefaultAgent()
|
||||
|
||||
if agent == nil {
|
||||
return ""
|
||||
}
|
||||
|
||||
prompt := agent.ContextBuilder.BuildSystemPrompt()
|
||||
|
||||
al.lastSystemPrompt.Store(prompt)
|
||||
|
||||
al.promptDirty.Store(false)
|
||||
|
||||
return prompt
|
||||
}
|
||||
|
||||
// formatMessagesForLog formats messages for logging
|
||||
|
||||
func formatMessagesForLog(messages []providers.Message) string {
|
||||
if len(messages) == 0 {
|
||||
return "[]"
|
||||
}
|
||||
|
||||
var sb strings.Builder
|
||||
|
||||
sb.WriteString("[\n")
|
||||
|
||||
for i, msg := range messages {
|
||||
fmt.Fprintf(&sb, " [%d] Role: %s\n", i, msg.Role)
|
||||
|
||||
if len(msg.ToolCalls) > 0 {
|
||||
sb.WriteString(" ToolCalls:\n")
|
||||
|
||||
for _, tc := range msg.ToolCalls {
|
||||
fmt.Fprintf(&sb, " - ID: %s, Type: %s, Name: %s\n", tc.ID, tc.Type, tc.Name)
|
||||
|
||||
args := tc.Arguments
|
||||
|
||||
if len(args) == 0 && tc.Function != nil {
|
||||
args = tc.Function.Arguments
|
||||
}
|
||||
|
||||
if len(args) > 0 {
|
||||
argsJSON, _ := json.Marshal(args)
|
||||
|
||||
fmt.Fprintf(&sb, " Arguments: %s\n", utils.Truncate(string(argsJSON), 200))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if msg.Content != "" {
|
||||
content := utils.Truncate(msg.Content, 200)
|
||||
|
||||
fmt.Fprintf(&sb, " Content: %s\n", content)
|
||||
}
|
||||
|
||||
if msg.ToolCallID != "" {
|
||||
fmt.Fprintf(&sb, " ToolCallID: %s\n", msg.ToolCallID)
|
||||
}
|
||||
|
||||
sb.WriteString("\n")
|
||||
}
|
||||
|
||||
sb.WriteString("]")
|
||||
|
||||
return sb.String()
|
||||
}
|
||||
|
||||
// formatToolsForLog formats tool definitions for logging
|
||||
|
||||
func formatToolsForLog(toolDefs []providers.ToolDefinition) string {
|
||||
if len(toolDefs) == 0 {
|
||||
return "[]"
|
||||
}
|
||||
|
||||
var sb strings.Builder
|
||||
|
||||
sb.WriteString("[\n")
|
||||
|
||||
for i, tool := range toolDefs {
|
||||
fmt.Fprintf(&sb, " [%d] Type: %s, Name: %s\n", i, tool.Type, tool.Function.Name)
|
||||
|
||||
fmt.Fprintf(&sb, " Description: %s\n", tool.Function.Description)
|
||||
|
||||
if len(tool.Function.Parameters) > 0 {
|
||||
fmt.Fprintf(&sb, " Parameters: %s\n", utils.Truncate(string(tool.Function.Parameters), 200))
|
||||
}
|
||||
}
|
||||
|
||||
sb.WriteString("]")
|
||||
|
||||
return sb.String()
|
||||
}
|
||||
340
pkg/agent/loop_orch.go
Normal file
340
pkg/agent/loop_orch.go
Normal file
|
|
@ -0,0 +1,340 @@
|
|||
package agent
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"github.com/sipeed/picoclaw/pkg/bus"
|
||||
"github.com/sipeed/picoclaw/pkg/constants"
|
||||
"github.com/sipeed/picoclaw/pkg/logger"
|
||||
"github.com/sipeed/picoclaw/pkg/orch"
|
||||
"github.com/sipeed/picoclaw/pkg/providers"
|
||||
"github.com/sipeed/picoclaw/pkg/routing"
|
||||
"github.com/sipeed/picoclaw/pkg/session"
|
||||
)
|
||||
|
||||
func (al *AgentLoop) reporter() orch.AgentReporter {
|
||||
if al.orchReporter == nil {
|
||||
return orch.Noop
|
||||
}
|
||||
|
||||
return al.orchReporter
|
||||
}
|
||||
|
||||
// SetOrchReporter wires a Broadcaster as the active reporter.
|
||||
|
||||
// Called from cmd_gateway.go when --orchestration is set.
|
||||
|
||||
// --orchestration なし → 呼ばれない → reporter() は Noop を返す。
|
||||
|
||||
func (al *AgentLoop) SetOrchReporter(b *orch.Broadcaster) {
|
||||
al.orchBroadcaster = b
|
||||
|
||||
al.orchReporter = b
|
||||
}
|
||||
|
||||
// GetOrchBroadcaster returns the concrete Broadcaster for miniapp wiring.
|
||||
|
||||
// Returns nil when orchestration is disabled.
|
||||
|
||||
func (al *AgentLoop) GetOrchBroadcaster() *orch.Broadcaster {
|
||||
return al.orchBroadcaster
|
||||
}
|
||||
|
||||
func (al *AgentLoop) notifyStateChange() {
|
||||
al.promptDirty.Store(true)
|
||||
|
||||
if al.OnStateChange != nil {
|
||||
al.OnStateChange()
|
||||
}
|
||||
}
|
||||
|
||||
func (al *AgentLoop) processSystemMessage(ctx context.Context, msg bus.InboundMessage) (string, error) {
|
||||
if msg.Channel != "system" {
|
||||
return "", fmt.Errorf("processSystemMessage called with non-system message channel: %s", msg.Channel)
|
||||
}
|
||||
|
||||
logger.InfoCF("agent", "Processing system message",
|
||||
|
||||
map[string]any{
|
||||
"sender_id": msg.SenderID,
|
||||
|
||||
"chat_id": msg.ChatID,
|
||||
})
|
||||
|
||||
// Parse origin channel from chat_id (format: "channel:chat_id")
|
||||
|
||||
var originChannel, originChatID string
|
||||
|
||||
if idx := strings.Index(msg.ChatID, ":"); idx > 0 {
|
||||
originChannel = msg.ChatID[:idx]
|
||||
|
||||
originChatID = msg.ChatID[idx+1:]
|
||||
} else {
|
||||
originChannel = "cli"
|
||||
|
||||
originChatID = msg.ChatID
|
||||
}
|
||||
|
||||
// Extract subagent result from message content
|
||||
|
||||
// Format: "Task 'label' completed.\n\nResult:\n<actual content>"
|
||||
|
||||
content := msg.Content
|
||||
|
||||
if idx := strings.Index(content, "Result:\n"); idx >= 0 {
|
||||
content = content[idx+8:] // Extract just the result part
|
||||
}
|
||||
|
||||
// Skip internal channels - only log, don't send to user
|
||||
|
||||
if constants.IsInternalChannel(originChannel) {
|
||||
logger.InfoCF("agent", "Subagent completed (internal channel)",
|
||||
|
||||
map[string]any{
|
||||
"sender_id": msg.SenderID,
|
||||
|
||||
"content_len": len(content),
|
||||
|
||||
"channel": originChannel,
|
||||
})
|
||||
|
||||
return "", nil
|
||||
}
|
||||
|
||||
// Inject subagent result into session history without running a full LLM loop.
|
||||
|
||||
// The conductor will see the result on its next turn. This avoids:
|
||||
|
||||
// - Flooding the chat with a response for every subagent completion
|
||||
|
||||
// - Consuming the Telegram "Thinking..." placeholder
|
||||
|
||||
// - Wasting LLM tokens on processing each result individually
|
||||
|
||||
agent := al.registry.GetDefaultAgent()
|
||||
|
||||
if agent == nil {
|
||||
return "", fmt.Errorf("no default agent for system message")
|
||||
}
|
||||
|
||||
sessionKey := routing.BuildAgentMainSessionKey(agent.ID)
|
||||
|
||||
historyMsg := fmt.Sprintf("[System: %s] %s", msg.SenderID, msg.Content)
|
||||
|
||||
// Write as TurnReport to the store for DAG tracking, with legacy fallback.
|
||||
|
||||
subagentSessionKey := routing.BuildSubagentSessionKey(extractTaskID(msg.SenderID))
|
||||
|
||||
store := agent.Sessions.Store()
|
||||
|
||||
reportTurn := &session.Turn{
|
||||
Kind: session.TurnReport,
|
||||
|
||||
OriginKey: subagentSessionKey,
|
||||
|
||||
Author: msg.SenderID,
|
||||
|
||||
Messages: []providers.Message{{Role: "user", Content: historyMsg}},
|
||||
}
|
||||
|
||||
if err := store.Append(sessionKey, reportTurn); err != nil {
|
||||
logger.ErrorCF("agent", "Failed to record report turn, falling back to legacy",
|
||||
|
||||
map[string]any{"error": err.Error()})
|
||||
|
||||
agent.Sessions.AddMessage(sessionKey, "user", historyMsg)
|
||||
|
||||
agent.Sessions.MarkDirty(sessionKey)
|
||||
} else {
|
||||
// Update in-memory cache so conductor sees the message on next turn.
|
||||
|
||||
agent.Sessions.AddFullMessage(sessionKey, providers.Message{Role: "user", Content: historyMsg})
|
||||
|
||||
agent.Sessions.AdvanceStored(sessionKey, 1)
|
||||
}
|
||||
|
||||
// Send a brief notification (SkipPlaceholder to avoid corrupting status messages)
|
||||
|
||||
label := msg.SenderID
|
||||
|
||||
if idx := strings.LastIndex(label, ":"); idx >= 0 {
|
||||
label = label[idx+1:]
|
||||
}
|
||||
|
||||
notification := formatSubagentCompletion(label, msg.Metadata)
|
||||
|
||||
subagentThreadID := 0
|
||||
|
||||
if al.cfg != nil {
|
||||
subagentThreadID = al.cfg.Channels.Telegram.SubagentThreadID
|
||||
}
|
||||
|
||||
notifyChatID := al.withTelegramThread(originChannel, originChatID, subagentThreadID)
|
||||
|
||||
_ = al.bus.PublishOutbound(ctx, bus.OutboundMessage{
|
||||
Channel: originChannel,
|
||||
|
||||
ChatID: notifyChatID,
|
||||
|
||||
Content: notification,
|
||||
|
||||
SkipPlaceholder: true,
|
||||
})
|
||||
|
||||
logger.InfoCF("agent", "Subagent result injected into session history",
|
||||
|
||||
map[string]any{
|
||||
"sender_id": msg.SenderID,
|
||||
|
||||
"session_key": sessionKey,
|
||||
|
||||
"content_len": len(content),
|
||||
})
|
||||
|
||||
return "", nil
|
||||
}
|
||||
|
||||
// extractTaskID extracts the task ID from a sender ID like "subagent:subagent-1".
|
||||
|
||||
func extractTaskID(senderID string) string {
|
||||
if idx := strings.LastIndex(senderID, ":"); idx >= 0 {
|
||||
return senderID[idx+1:]
|
||||
}
|
||||
|
||||
return senderID
|
||||
}
|
||||
|
||||
// formatSubagentCompletion builds the user-facing notification for a completed subagent.
|
||||
|
||||
// If metadata contains duration_ms and tool_calls it produces e.g.:
|
||||
|
||||
//
|
||||
|
||||
// "📋 scout-1 completed (3.2s, 5 tool calls)."
|
||||
|
||||
//
|
||||
|
||||
// Without metadata it falls back to the plain "📋 scout-1 completed." format.
|
||||
|
||||
func formatSubagentCompletion(label string, metadata map[string]string) string {
|
||||
if len(metadata) == 0 {
|
||||
return fmt.Sprintf("📋 %s completed.", label)
|
||||
}
|
||||
|
||||
durationMs, _ := strconv.ParseInt(metadata["duration_ms"], 10, 64)
|
||||
|
||||
toolCalls, _ := strconv.Atoi(metadata["tool_calls"])
|
||||
|
||||
if durationMs <= 0 && toolCalls <= 0 {
|
||||
return fmt.Sprintf("📋 %s completed.", label)
|
||||
}
|
||||
|
||||
parts := make([]string, 0, 2)
|
||||
|
||||
if durationMs > 0 {
|
||||
parts = append(parts, formatDurationMs(durationMs))
|
||||
}
|
||||
|
||||
if toolCalls > 0 {
|
||||
if toolCalls == 1 {
|
||||
parts = append(parts, "1 tool call")
|
||||
} else {
|
||||
parts = append(parts, fmt.Sprintf("%d tool calls", toolCalls))
|
||||
}
|
||||
}
|
||||
|
||||
return fmt.Sprintf("📋 %s completed (%s).", label, strings.Join(parts, ", "))
|
||||
}
|
||||
|
||||
// formatDurationMs converts milliseconds to a human-readable duration string.
|
||||
|
||||
// Examples: 800 → "0.8s", 1200 → "1.2s", 65000 → "1m5s", 3661000 → "61m1s".
|
||||
|
||||
func formatDurationMs(ms int64) string {
|
||||
if ms < 1000 {
|
||||
return fmt.Sprintf("%dms", ms)
|
||||
}
|
||||
|
||||
totalSec := ms / 1000
|
||||
|
||||
if totalSec < 60 {
|
||||
tenths := (ms % 1000) / 100
|
||||
|
||||
return fmt.Sprintf("%d.%ds", totalSec, tenths)
|
||||
}
|
||||
|
||||
mins := totalSec / 60
|
||||
|
||||
sec := totalSec % 60
|
||||
|
||||
if sec == 0 {
|
||||
return fmt.Sprintf("%dm", mins)
|
||||
}
|
||||
|
||||
return fmt.Sprintf("%dm%ds", mins, sec)
|
||||
}
|
||||
|
||||
// buildOrchReminder returns a reminder to use spawn/subagent during plan execution.
|
||||
|
||||
// Fires on first iteration and every 3rd iteration to reinforce delegation behavior.
|
||||
|
||||
func buildOrchReminder(iteration int) (providers.Message, bool) {
|
||||
if iteration != 1 && iteration%3 != 0 {
|
||||
return providers.Message{}, false
|
||||
}
|
||||
|
||||
content := `[System] ORCHESTRATION mode active. You MUST delegate plan steps to subagents.
|
||||
|
||||
Use spawn (non-blocking, returns immediately) or subagent (blocking, waits for result).
|
||||
|
||||
Do NOT implement steps inline unless they are a single trivial tool call.
|
||||
|
||||
|
||||
|
||||
To delegate, call the tool with JSON arguments:
|
||||
|
||||
Tool: spawn Arguments: {"task": "...", "preset": "scout", "label": "..."}
|
||||
|
||||
Tool: subagent Arguments: {"task": "...", "label": "..."}
|
||||
|
||||
|
||||
|
||||
Spawn multiple independent steps in parallel for maximum throughput.`
|
||||
|
||||
return providers.Message{Role: "user", Content: content}, true
|
||||
}
|
||||
|
||||
func extractPeer(msg bus.InboundMessage) *routing.RoutePeer {
|
||||
if msg.Peer.Kind == "" {
|
||||
return nil
|
||||
}
|
||||
|
||||
peerID := msg.Peer.ID
|
||||
|
||||
if peerID == "" {
|
||||
if msg.Peer.Kind == "direct" {
|
||||
peerID = msg.SenderID
|
||||
} else {
|
||||
peerID = msg.ChatID
|
||||
}
|
||||
}
|
||||
|
||||
return &routing.RoutePeer{Kind: msg.Peer.Kind, ID: peerID}
|
||||
}
|
||||
|
||||
// extractParentPeer extracts the parent peer (reply-to) from inbound message metadata.
|
||||
|
||||
func extractParentPeer(msg bus.InboundMessage) *routing.RoutePeer {
|
||||
parentKind := msg.Metadata["parent_peer_kind"]
|
||||
|
||||
parentID := msg.Metadata["parent_peer_id"]
|
||||
|
||||
if parentKind == "" || parentID == "" {
|
||||
return nil
|
||||
}
|
||||
|
||||
return &routing.RoutePeer{Kind: parentKind, ID: parentID}
|
||||
}
|
||||
674
pkg/agent/loop_plan.go
Normal file
674
pkg/agent/loop_plan.go
Normal file
|
|
@ -0,0 +1,674 @@
|
|||
package agent
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"github.com/sipeed/picoclaw/pkg/bus"
|
||||
"github.com/sipeed/picoclaw/pkg/git"
|
||||
"github.com/sipeed/picoclaw/pkg/orch"
|
||||
"github.com/sipeed/picoclaw/pkg/providers"
|
||||
"github.com/sipeed/picoclaw/pkg/tools"
|
||||
"github.com/sipeed/picoclaw/pkg/utils"
|
||||
)
|
||||
|
||||
// interviewRejectMessage is the fixed rejection text injected when tool calls
|
||||
|
||||
// are blocked during the interview phase. It is deliberately short to avoid
|
||||
|
||||
// wasting tokens, and ends with a purpose reminder to steer the LLM back.
|
||||
|
||||
const interviewRejectMessage = "[System] Tool call rejected. " +
|
||||
|
||||
"You are in interview mode — ask the user questions and update MEMORY.md. " +
|
||||
|
||||
"Do not execute, edit, or write project files."
|
||||
|
||||
// buildPlanReminder returns a reminder message for plan pre-execution states
|
||||
|
||||
// (interviewing / review) to keep the AI focused on the interview workflow
|
||||
|
||||
// during tool-call iterations.
|
||||
|
||||
func buildPlanReminder(planStatus string) (providers.Message, bool) {
|
||||
var content string
|
||||
|
||||
switch planStatus {
|
||||
case "interviewing":
|
||||
|
||||
content = "[System] You are interviewing the user to build a plan. " +
|
||||
|
||||
"Ask clarifying questions and save findings to ## Context in memory/MEMORY.md using edit_file. " +
|
||||
|
||||
"When you have enough information, write ## Phase sections with `- [ ]` checkbox steps, and ## Commands section. " +
|
||||
|
||||
"Then change > Status: to review. Do NOT set it to executing."
|
||||
|
||||
case "review":
|
||||
|
||||
content = "[System] The plan is under review. " +
|
||||
|
||||
"Wait for the user to approve or request changes. Do not proceed with execution."
|
||||
|
||||
default:
|
||||
|
||||
return providers.Message{}, false
|
||||
}
|
||||
|
||||
return providers.Message{Role: "user", Content: content}, true
|
||||
}
|
||||
|
||||
func (al *AgentLoop) handlePlanCommand(args []string, sessionKey string) (string, bool) {
|
||||
agent := al.registry.GetDefaultAgent()
|
||||
|
||||
if agent == nil {
|
||||
return "No agent configured.", true
|
||||
}
|
||||
|
||||
if len(args) == 0 {
|
||||
// /plan — show current plan
|
||||
|
||||
return agent.ContextBuilder.FormatPlanDisplay(), true
|
||||
}
|
||||
|
||||
sub := args[0]
|
||||
|
||||
switch sub {
|
||||
case "clear":
|
||||
|
||||
if agent.ContextBuilder.ReadMemory() == "" {
|
||||
return "No active plan to clear.", true
|
||||
}
|
||||
|
||||
// Deactivate worktree on plan clear
|
||||
|
||||
if sessionKey != "" {
|
||||
agent.DeactivateWorktree(sessionKey, "", true)
|
||||
}
|
||||
|
||||
if err := agent.ContextBuilder.ClearMemory(); err != nil {
|
||||
return fmt.Sprintf("Error clearing plan: %v", err), true
|
||||
}
|
||||
|
||||
return "Plan cleared.", true
|
||||
|
||||
case "done":
|
||||
|
||||
if !agent.ContextBuilder.HasActivePlan() {
|
||||
return "No active plan.", true
|
||||
}
|
||||
|
||||
if len(args) < 2 {
|
||||
return "Usage: /plan done <step number>", true
|
||||
}
|
||||
|
||||
stepNum, err := strconv.Atoi(args[1])
|
||||
|
||||
if err != nil || stepNum < 1 {
|
||||
return "Step number must be a positive integer.", true
|
||||
}
|
||||
|
||||
phase := agent.ContextBuilder.GetCurrentPhase()
|
||||
|
||||
if err := agent.ContextBuilder.MarkStep(phase, stepNum); err != nil {
|
||||
return fmt.Sprintf("Error: %v", err), true
|
||||
}
|
||||
|
||||
return fmt.Sprintf("Marked step %d in phase %d as done.", stepNum, phase), true
|
||||
|
||||
case "add":
|
||||
|
||||
if !agent.ContextBuilder.HasActivePlan() {
|
||||
return "No active plan.", true
|
||||
}
|
||||
|
||||
if len(args) < 2 {
|
||||
return "Usage: /plan add <step description>", true
|
||||
}
|
||||
|
||||
desc := strings.Join(args[1:], " ")
|
||||
|
||||
phase := agent.ContextBuilder.GetCurrentPhase()
|
||||
|
||||
if err := agent.ContextBuilder.AddStep(phase, desc); err != nil {
|
||||
return fmt.Sprintf("Error: %v", err), true
|
||||
}
|
||||
|
||||
return fmt.Sprintf("Added step to phase %d: %s", phase, desc), true
|
||||
|
||||
case "start":
|
||||
|
||||
if !agent.ContextBuilder.HasActivePlan() {
|
||||
return "No active plan.", true
|
||||
}
|
||||
|
||||
status := agent.ContextBuilder.GetPlanStatus()
|
||||
|
||||
if status == "executing" {
|
||||
return "Plan is already executing.", true
|
||||
}
|
||||
|
||||
if status != "interviewing" && status != "review" {
|
||||
return fmt.Sprintf("Cannot start from status %q.", status), true
|
||||
}
|
||||
|
||||
if agent.ContextBuilder.GetTotalPhases() == 0 {
|
||||
return "Cannot start: no phases defined yet. Complete the interview first.", true
|
||||
}
|
||||
|
||||
if err := agent.ContextBuilder.SetPlanStatus("executing"); err != nil {
|
||||
return fmt.Sprintf("Error: %v", err), true
|
||||
}
|
||||
|
||||
al.reporter().ReportStateChange(sessionKey, orch.AgentStatePlanExecuting, "")
|
||||
|
||||
al.planStartPending = true
|
||||
|
||||
clearHistory := len(args) > 1 && args[1] == "clear"
|
||||
|
||||
al.planClearHistory = clearHistory
|
||||
|
||||
if clearHistory {
|
||||
return "Plan approved. Executing with clean history.", true
|
||||
}
|
||||
|
||||
return "Plan approved. Executing.", true
|
||||
|
||||
case "next":
|
||||
|
||||
if !agent.ContextBuilder.HasActivePlan() {
|
||||
return "No active plan.", true
|
||||
}
|
||||
|
||||
if err := agent.ContextBuilder.AdvancePhase(); err != nil {
|
||||
return fmt.Sprintf("Error: %v", err), true
|
||||
}
|
||||
|
||||
phase := agent.ContextBuilder.GetCurrentPhase()
|
||||
|
||||
return fmt.Sprintf("Advanced to phase %d.", phase), true
|
||||
|
||||
case "worktrees":
|
||||
|
||||
return al.handlePlanWorktreesCommand(agent, args[1:]), true
|
||||
|
||||
default:
|
||||
|
||||
// /plan <task description> — start new plan
|
||||
|
||||
// Block if a plan is already active (fast-path error).
|
||||
|
||||
if agent.ContextBuilder.HasActivePlan() {
|
||||
return "A plan is already active. Use /plan clear first.", true
|
||||
}
|
||||
|
||||
// Not handled here — let the message flow to the LLM queue.
|
||||
|
||||
// expandPlanCommand will write the seed and rewrite the content.
|
||||
|
||||
return "", false
|
||||
}
|
||||
}
|
||||
|
||||
func (al *AgentLoop) handlePlanWorktreesCommand(agent *AgentInstance, args []string) string {
|
||||
repoRoot := git.FindRepoRoot(agent.Workspace)
|
||||
|
||||
if repoRoot == "" {
|
||||
return "Workspace is not a git repository."
|
||||
}
|
||||
|
||||
worktreesDir := filepath.Join(agent.Workspace, ".worktrees")
|
||||
|
||||
sub := "list"
|
||||
|
||||
if len(args) > 0 {
|
||||
sub = strings.ToLower(strings.TrimSpace(args[0]))
|
||||
}
|
||||
|
||||
switch sub {
|
||||
case "", "list":
|
||||
|
||||
items, err := git.ListManagedWorktrees(repoRoot, worktreesDir)
|
||||
if err != nil {
|
||||
return fmt.Sprintf("Error listing worktrees: %v", err)
|
||||
}
|
||||
|
||||
if len(items) == 0 {
|
||||
return "No active worktrees in workspace/.worktrees."
|
||||
}
|
||||
|
||||
var sb strings.Builder
|
||||
|
||||
sb.WriteString("Active worktrees\n\n")
|
||||
|
||||
for _, wt := range items {
|
||||
status := "clean"
|
||||
|
||||
if wt.HasUncommitted {
|
||||
status = "dirty"
|
||||
}
|
||||
|
||||
last := "(no commits)"
|
||||
|
||||
if wt.LastCommitHash != "" {
|
||||
if wt.LastCommitAge != "" {
|
||||
last = fmt.Sprintf("%s %s (%s)", wt.LastCommitHash, wt.LastCommitSubject, wt.LastCommitAge)
|
||||
} else {
|
||||
last = fmt.Sprintf("%s %s", wt.LastCommitHash, wt.LastCommitSubject)
|
||||
}
|
||||
}
|
||||
|
||||
fmt.Fprintf(&sb, "- %s\n branch: %s\n status: %s\n last: %s\n", wt.Name, wt.Branch, status, last)
|
||||
}
|
||||
|
||||
sb.WriteString("\nCommands:\n")
|
||||
|
||||
sb.WriteString("/plan worktrees inspect <name>\n")
|
||||
|
||||
sb.WriteString("/plan worktrees merge <name>\n")
|
||||
|
||||
sb.WriteString("/plan worktrees dispose <name> [force]")
|
||||
|
||||
return sb.String()
|
||||
|
||||
case "inspect":
|
||||
|
||||
if len(args) < 2 {
|
||||
return "Usage: /plan worktrees inspect <name>"
|
||||
}
|
||||
|
||||
name := args[1]
|
||||
|
||||
wt, err := git.GetManagedWorktree(repoRoot, worktreesDir, name)
|
||||
if err != nil {
|
||||
if errors.Is(err, git.ErrInvalidWorktreeName) {
|
||||
return "Invalid worktree name."
|
||||
}
|
||||
|
||||
if errors.Is(err, git.ErrWorktreeNotFound) {
|
||||
return fmt.Sprintf("Worktree %q not found.", name)
|
||||
}
|
||||
|
||||
return fmt.Sprintf("Error inspecting worktree %q: %v", name, err)
|
||||
}
|
||||
|
||||
statusOut, _ := git.WorktreeStatusShort(wt.Path)
|
||||
|
||||
diffOut, _ := git.WorktreeDiffStat(wt.Path)
|
||||
|
||||
logOut, _ := git.WorktreeRecentLog(wt.Path, 10)
|
||||
|
||||
if statusOut == "" {
|
||||
statusOut = "(clean)"
|
||||
}
|
||||
|
||||
var sb strings.Builder
|
||||
|
||||
fmt.Fprintf(&sb, "Worktree: %s\nBranch: %s\nDirty: %t\n", wt.Name, wt.Branch, wt.HasUncommitted)
|
||||
|
||||
if wt.LastCommitHash != "" {
|
||||
fmt.Fprintf(&sb, "Last commit: %s %s", wt.LastCommitHash, wt.LastCommitSubject)
|
||||
|
||||
if wt.LastCommitAge != "" {
|
||||
fmt.Fprintf(&sb, " (%s)", wt.LastCommitAge)
|
||||
}
|
||||
|
||||
sb.WriteString("\n")
|
||||
}
|
||||
|
||||
sb.WriteString("\nStatus:\n```\n")
|
||||
|
||||
sb.WriteString(statusOut)
|
||||
|
||||
sb.WriteString("\n```\n")
|
||||
|
||||
if diffOut != "" {
|
||||
sb.WriteString("\nDiff (stat):\n```\n")
|
||||
|
||||
sb.WriteString(diffOut)
|
||||
|
||||
sb.WriteString("\n```\n")
|
||||
}
|
||||
|
||||
if logOut != "" {
|
||||
sb.WriteString("\nRecent commits:\n```\n")
|
||||
|
||||
sb.WriteString(logOut)
|
||||
|
||||
sb.WriteString("\n```")
|
||||
}
|
||||
|
||||
return sb.String()
|
||||
|
||||
case "merge":
|
||||
|
||||
if len(args) < 2 {
|
||||
return "Usage: /plan worktrees merge <name>"
|
||||
}
|
||||
|
||||
name := args[1]
|
||||
|
||||
res, base, err := git.MergeManagedWorktree(repoRoot, worktreesDir, name, "")
|
||||
if err != nil {
|
||||
if errors.Is(err, git.ErrInvalidWorktreeName) {
|
||||
return "Invalid worktree name."
|
||||
}
|
||||
|
||||
if errors.Is(err, git.ErrWorktreeNotFound) {
|
||||
return fmt.Sprintf("Worktree %q not found.", name)
|
||||
}
|
||||
|
||||
return fmt.Sprintf("Error merging worktree %q: %v", name, err)
|
||||
}
|
||||
|
||||
if res.Conflict {
|
||||
return fmt.Sprintf("Merge conflict while merging `%s` into `%s`. Merge was aborted.", res.Branch, base)
|
||||
}
|
||||
|
||||
if res.Merged {
|
||||
return fmt.Sprintf("Merged `%s` into `%s`.", res.Branch, base)
|
||||
}
|
||||
|
||||
return fmt.Sprintf("No merge was performed for `%s`.", name)
|
||||
|
||||
case "dispose":
|
||||
|
||||
if len(args) < 2 {
|
||||
return "Usage: /plan worktrees dispose <name> [force]"
|
||||
}
|
||||
|
||||
name := args[1]
|
||||
|
||||
force := len(args) > 2 && strings.EqualFold(args[2], "force")
|
||||
|
||||
wt, err := git.GetManagedWorktree(repoRoot, worktreesDir, name)
|
||||
if err != nil {
|
||||
if errors.Is(err, git.ErrInvalidWorktreeName) {
|
||||
return "Invalid worktree name."
|
||||
}
|
||||
|
||||
if errors.Is(err, git.ErrWorktreeNotFound) {
|
||||
return fmt.Sprintf("Worktree %q not found.", name)
|
||||
}
|
||||
|
||||
return fmt.Sprintf("Error disposing worktree %q: %v", name, err)
|
||||
}
|
||||
|
||||
if wt.HasUncommitted && !force {
|
||||
return fmt.Sprintf(
|
||||
|
||||
"Worktree `%s` has uncommitted changes. Re-run with `/plan worktrees dispose %s force` to confirm.",
|
||||
|
||||
name,
|
||||
|
||||
name,
|
||||
)
|
||||
}
|
||||
|
||||
res, err := git.DisposeManagedWorktree(repoRoot, worktreesDir, name, "")
|
||||
if err != nil {
|
||||
return fmt.Sprintf("Error disposing worktree %q: %v", name, err)
|
||||
}
|
||||
|
||||
parts := []string{fmt.Sprintf("Disposed worktree `%s` (branch `%s`).", name, res.Branch)}
|
||||
|
||||
if res.AutoCommitted {
|
||||
parts = append(parts, "Uncommitted changes were auto-committed.")
|
||||
}
|
||||
|
||||
if res.CommitsAhead > 0 {
|
||||
parts = append(parts, fmt.Sprintf("Branch has %d unique commit(s); branch was kept.", res.CommitsAhead))
|
||||
}
|
||||
|
||||
if res.BranchDeleted {
|
||||
parts = append(parts, "Branch was deleted (no unique commits).")
|
||||
}
|
||||
|
||||
return strings.Join(parts, " ")
|
||||
}
|
||||
|
||||
return "Usage: /plan worktrees [list|inspect <name>|merge <name>|dispose <name> [force]]"
|
||||
}
|
||||
|
||||
// isPlanPreExecution returns true if the plan is in a pre-execution state
|
||||
|
||||
// (interviewing or review) where tool restrictions and iteration caps apply.
|
||||
|
||||
func isPlanPreExecution(status string) bool {
|
||||
return status == "interviewing" || status == "review"
|
||||
}
|
||||
|
||||
// interviewAllowedTools is the single source of truth for tool names that may
|
||||
|
||||
// be sent to the LLM (and subsequently invoked) during the interview phase.
|
||||
|
||||
// filterInterviewTools uses this to strip tool *definitions* before the LLM call,
|
||||
|
||||
// while isToolAllowedDuringInterview adds argument-level checks as a second gate.
|
||||
|
||||
var interviewAllowedTools = map[string]bool{
|
||||
"readfile": true,
|
||||
|
||||
"listdir": true,
|
||||
|
||||
"websearch": true,
|
||||
|
||||
"webfetch": true,
|
||||
|
||||
"message": true,
|
||||
|
||||
"editfile": true,
|
||||
|
||||
"appendfile": true,
|
||||
|
||||
"writefile": true,
|
||||
|
||||
"exec": true,
|
||||
|
||||
"logs": true,
|
||||
}
|
||||
|
||||
// filterInterviewTools removes tool definitions that are not in the
|
||||
|
||||
// interviewAllowedTools whitelist, reducing token usage and preventing the
|
||||
|
||||
// LLM from attempting disallowed tool calls during the interview phase.
|
||||
|
||||
func filterInterviewTools(defs []providers.ToolDefinition) []providers.ToolDefinition {
|
||||
filtered := make([]providers.ToolDefinition, 0, len(defs))
|
||||
|
||||
for _, d := range defs {
|
||||
if interviewAllowedTools[tools.NormalizeToolName(d.Function.Name)] {
|
||||
filtered = append(filtered, d)
|
||||
}
|
||||
}
|
||||
|
||||
return filtered
|
||||
}
|
||||
|
||||
// isToolAllowedDuringInterview checks whether a tool call is permitted while the
|
||||
|
||||
// plan is in a pre-execution state. Uses the shared interviewAllowedTools map for
|
||||
|
||||
// name-level gating, then applies argument-level constraints for write-type tools
|
||||
|
||||
// (MEMORY.md only) and exec (read-only commands only).
|
||||
|
||||
func isToolAllowedDuringInterview(toolName string, args map[string]any) bool {
|
||||
norm := tools.NormalizeToolName(toolName)
|
||||
|
||||
if !interviewAllowedTools[norm] {
|
||||
return false
|
||||
}
|
||||
|
||||
// Argument-level constraints
|
||||
|
||||
switch norm {
|
||||
case "editfile", "appendfile", "writefile":
|
||||
|
||||
path, _ := args["path"].(string)
|
||||
|
||||
return strings.HasSuffix(path, "MEMORY.md")
|
||||
|
||||
case "exec":
|
||||
|
||||
cmd, _ := args["command"].(string)
|
||||
|
||||
return isReadOnlyCommand(cmd)
|
||||
}
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
// isReadOnlyCommand returns true when cmd is a safe, read-only shell command
|
||||
|
||||
// that an LLM may run during the interview phase.
|
||||
|
||||
func isReadOnlyCommand(cmd string) bool {
|
||||
cmd = strings.TrimSpace(cmd)
|
||||
|
||||
if cmd == "" {
|
||||
return false
|
||||
}
|
||||
|
||||
// Reject write operators anywhere in the command
|
||||
|
||||
for _, op := range []string{">", ">>", "| tee "} {
|
||||
if strings.Contains(cmd, op) {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
// Reject path traversal (defense in depth; ExecTool.guardCommand also enforces workspace restriction)
|
||||
|
||||
if strings.Contains(cmd, "..") {
|
||||
return false
|
||||
}
|
||||
|
||||
// Block absolute paths in arguments (allow "cd /path && cmd" which is stripped later)
|
||||
|
||||
for _, field := range strings.Fields(cmd) {
|
||||
if strings.HasPrefix(field, "/") && !strings.HasPrefix(cmd, "cd ") {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
// Strip "cd /path &&" prefix (LLM habit)
|
||||
|
||||
if strings.HasPrefix(cmd, "cd ") {
|
||||
if idx := strings.Index(cmd, "&&"); idx >= 0 {
|
||||
cmd = strings.TrimSpace(cmd[idx+2:])
|
||||
}
|
||||
}
|
||||
|
||||
fields := strings.Fields(cmd)
|
||||
|
||||
if len(fields) == 0 {
|
||||
return false
|
||||
}
|
||||
|
||||
first := filepath.Base(fields[0])
|
||||
|
||||
switch first {
|
||||
case "find", "ls", "cat", "head", "tail", "grep", "rg",
|
||||
|
||||
"tree", "wc", "file", "which", "pwd",
|
||||
|
||||
"uname", "df", "du", "stat", "realpath", "dirname",
|
||||
|
||||
"basename", "date":
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
// isWriteTool returns true if the tool can modify files.
|
||||
|
||||
func isWriteTool(name string) bool {
|
||||
switch tools.NormalizeToolName(name) {
|
||||
case "writefile", "editfile", "appendfile", "exec":
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
// expandPlanCommand detects "/plan <task>" (new plan start) and:
|
||||
|
||||
// - writes the interview seed to MEMORY.md
|
||||
|
||||
// - rewrites the message content for the LLM
|
||||
|
||||
// - returns a compact form for session history
|
||||
|
||||
//
|
||||
|
||||
// This follows the same pattern as expandSkillCommand: the message is
|
||||
|
||||
// rewritten before reaching the LLM, so the AI sees the task description
|
||||
|
||||
// while the system prompt contains the interview guide.
|
||||
|
||||
func (al *AgentLoop) expandPlanCommand(msg bus.InboundMessage) (expanded string, compact string, ok bool) {
|
||||
content := strings.TrimSpace(msg.Content)
|
||||
|
||||
if !strings.HasPrefix(content, "/plan ") {
|
||||
return "", "", false
|
||||
}
|
||||
|
||||
task := strings.TrimSpace(content[6:]) // len("/plan ") == 6
|
||||
|
||||
if task == "" {
|
||||
return "", "", false
|
||||
}
|
||||
|
||||
// Known subcommands are handled by handlePlanCommand (fast path).
|
||||
|
||||
firstWord := strings.Fields(task)[0]
|
||||
|
||||
switch firstWord {
|
||||
case "clear", "done", "add", "start", "next", "worktrees":
|
||||
|
||||
return "", "", false
|
||||
}
|
||||
|
||||
agent := al.registry.GetDefaultAgent()
|
||||
|
||||
if agent == nil {
|
||||
return "", "", false
|
||||
}
|
||||
|
||||
// If a plan is already active, don't expand — handleCommand will
|
||||
|
||||
// catch it and return the error on the fast path.
|
||||
|
||||
if agent.ContextBuilder.HasActivePlan() {
|
||||
return "", "", false
|
||||
}
|
||||
|
||||
// Write the interview seed
|
||||
|
||||
seed := BuildInterviewSeed(task, agent.Workspace)
|
||||
|
||||
if err := agent.ContextBuilder.WriteMemory(seed); err != nil {
|
||||
return "", "", false
|
||||
}
|
||||
|
||||
al.notifyStateChange()
|
||||
|
||||
// Expanded: the task description goes to LLM.
|
||||
|
||||
// The system prompt already contains the interview guide.
|
||||
|
||||
expanded = task
|
||||
|
||||
compact = fmt.Sprintf("[Plan: %s]", utils.Truncate(task, 80))
|
||||
|
||||
return expanded, compact, true
|
||||
}
|
||||
393
pkg/agent/loop_session.go
Normal file
393
pkg/agent/loop_session.go
Normal file
|
|
@ -0,0 +1,393 @@
|
|||
package agent
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
"unicode/utf8"
|
||||
|
||||
"github.com/sipeed/picoclaw/pkg/logger"
|
||||
"github.com/sipeed/picoclaw/pkg/providers"
|
||||
)
|
||||
|
||||
// sessionSemaphore is a per-session mutex using a buffered channel.
|
||||
|
||||
type sessionSemaphore struct {
|
||||
ch chan struct{}
|
||||
}
|
||||
|
||||
func newSessionSemaphore() *sessionSemaphore {
|
||||
s := &sessionSemaphore{ch: make(chan struct{}, 1)}
|
||||
|
||||
s.ch <- struct{}{} // initially unlocked
|
||||
|
||||
return s
|
||||
}
|
||||
|
||||
func (al *AgentLoop) gcLoop() {
|
||||
ticker := time.NewTicker(30 * time.Minute)
|
||||
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
|
||||
al.gcSessionLocks()
|
||||
|
||||
case <-al.done:
|
||||
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// gcSessionLocks removes unlocked (idle) sessionSemaphore entries from the map.
|
||||
|
||||
func (al *AgentLoop) gcSessionLocks() {
|
||||
al.sessionLocks.Range(func(key, val any) bool {
|
||||
sem := val.(*sessionSemaphore)
|
||||
|
||||
select {
|
||||
case <-sem.ch:
|
||||
|
||||
// Was unlocked — safe to remove
|
||||
|
||||
al.sessionLocks.Delete(key)
|
||||
|
||||
default:
|
||||
|
||||
// Currently locked — in use, keep
|
||||
}
|
||||
|
||||
return true
|
||||
})
|
||||
}
|
||||
|
||||
func (al *AgentLoop) acquireSessionLock(ctx context.Context, sessionKey string) bool {
|
||||
val, _ := al.sessionLocks.LoadOrStore(sessionKey, newSessionSemaphore())
|
||||
|
||||
sem := val.(*sessionSemaphore)
|
||||
|
||||
select {
|
||||
case <-sem.ch:
|
||||
|
||||
return true
|
||||
|
||||
case <-ctx.Done():
|
||||
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
// releaseSessionLock releases the per-session semaphore.
|
||||
|
||||
func (al *AgentLoop) releaseSessionLock(sessionKey string) {
|
||||
if val, ok := al.sessionLocks.Load(sessionKey); ok {
|
||||
sem := val.(*sessionSemaphore)
|
||||
|
||||
sem.ch <- struct{}{}
|
||||
}
|
||||
}
|
||||
|
||||
func (al *AgentLoop) maybeSummarize(agent *AgentInstance, sessionKey, channel, chatID string) {
|
||||
newHistory := agent.Sessions.GetHistory(sessionKey)
|
||||
|
||||
tokenEstimate := al.estimateTokens(newHistory)
|
||||
|
||||
threshold := agent.ContextWindow * 75 / 100
|
||||
|
||||
if len(newHistory) > 20 || tokenEstimate > threshold {
|
||||
summarizeKey := agent.ID + ":" + sessionKey
|
||||
|
||||
if _, loading := al.summarizing.LoadOrStore(summarizeKey, true); !loading {
|
||||
go func() {
|
||||
defer al.summarizing.Delete(summarizeKey)
|
||||
|
||||
logger.InfoCF("agent", "Memory threshold reached, optimizing conversation history",
|
||||
|
||||
map[string]any{
|
||||
"session_key": sessionKey,
|
||||
|
||||
"history_len": len(newHistory),
|
||||
|
||||
"token_estimate": tokenEstimate,
|
||||
})
|
||||
|
||||
al.summarizeSession(agent, sessionKey)
|
||||
}()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// forceCompression aggressively reduces context when the limit is hit.
|
||||
|
||||
// It drops the oldest 50% of messages (keeping system prompt and last user message).
|
||||
|
||||
func (al *AgentLoop) forceCompression(agent *AgentInstance, sessionKey string) {
|
||||
history := agent.Sessions.GetHistory(sessionKey)
|
||||
|
||||
if len(history) <= 4 {
|
||||
return
|
||||
}
|
||||
|
||||
// Keep system prompt (usually [0]) and the very last message (user's trigger)
|
||||
|
||||
// We want to drop the oldest half of the *conversation*
|
||||
|
||||
// Assuming [0] is system, [1:] is conversation
|
||||
|
||||
conversation := history[1 : len(history)-1]
|
||||
|
||||
if len(conversation) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
// Helper to find the mid-point of the conversation
|
||||
|
||||
mid := len(conversation) / 2
|
||||
|
||||
// New history structure:
|
||||
|
||||
// 1. System Prompt (with compression note appended)
|
||||
|
||||
// 2. Second half of conversation
|
||||
|
||||
// 3. Last message
|
||||
|
||||
droppedCount := mid
|
||||
|
||||
keptConversation := conversation[mid:]
|
||||
|
||||
newHistory := make([]providers.Message, 0, 1+len(keptConversation)+1)
|
||||
|
||||
// Append compression note to the original system prompt instead of adding a new system message
|
||||
|
||||
// This avoids having two consecutive system messages which some APIs (like Zhipu) reject
|
||||
|
||||
compressionNote := fmt.Sprintf(
|
||||
|
||||
"\n\n[System Note: Emergency compression dropped %d oldest messages due to context limit]",
|
||||
|
||||
droppedCount,
|
||||
)
|
||||
|
||||
enhancedSystemPrompt := history[0]
|
||||
|
||||
enhancedSystemPrompt.Content = enhancedSystemPrompt.Content + compressionNote
|
||||
|
||||
newHistory = append(newHistory, enhancedSystemPrompt)
|
||||
|
||||
newHistory = append(newHistory, keptConversation...)
|
||||
|
||||
newHistory = append(newHistory, history[len(history)-1]) // Last message
|
||||
|
||||
// Update session
|
||||
|
||||
agent.Sessions.SetHistory(sessionKey, newHistory)
|
||||
|
||||
agent.Sessions.Save(sessionKey)
|
||||
|
||||
logger.WarnCF("agent", "Forced compression executed", map[string]any{
|
||||
"session_key": sessionKey,
|
||||
|
||||
"dropped_msgs": droppedCount,
|
||||
|
||||
"new_count": len(newHistory),
|
||||
})
|
||||
}
|
||||
|
||||
func (al *AgentLoop) summarizeSession(agent *AgentInstance, sessionKey string) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
|
||||
|
||||
defer cancel()
|
||||
|
||||
history := agent.Sessions.GetHistory(sessionKey)
|
||||
|
||||
summary := agent.Sessions.GetSummary(sessionKey)
|
||||
|
||||
// Keep last 4 messages for continuity
|
||||
|
||||
if len(history) <= 4 {
|
||||
return
|
||||
}
|
||||
|
||||
toSummarize := history[:len(history)-4]
|
||||
|
||||
// Oversized Message Guard
|
||||
|
||||
maxMessageTokens := agent.ContextWindow / 2
|
||||
|
||||
validMessages := make([]providers.Message, 0)
|
||||
|
||||
omitted := false
|
||||
|
||||
for _, m := range toSummarize {
|
||||
if m.Role != "user" && m.Role != "assistant" {
|
||||
continue
|
||||
}
|
||||
|
||||
msgTokens := len(m.Content) / 2
|
||||
|
||||
if msgTokens > maxMessageTokens {
|
||||
omitted = true
|
||||
|
||||
continue
|
||||
}
|
||||
|
||||
validMessages = append(validMessages, m)
|
||||
}
|
||||
|
||||
if len(validMessages) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
// Multi-Part Summarization
|
||||
|
||||
var finalSummary string
|
||||
|
||||
if len(validMessages) > 10 {
|
||||
mid := len(validMessages) / 2
|
||||
|
||||
part1 := validMessages[:mid]
|
||||
|
||||
part2 := validMessages[mid:]
|
||||
|
||||
s1, _ := al.summarizeBatch(ctx, agent, part1, "")
|
||||
|
||||
s2, _ := al.summarizeBatch(ctx, agent, part2, "")
|
||||
|
||||
mergePrompt := fmt.Sprintf(
|
||||
|
||||
"Merge these two conversation summaries into one cohesive summary:\n\n1: %s\n\n2: %s",
|
||||
|
||||
s1,
|
||||
|
||||
s2,
|
||||
)
|
||||
|
||||
resp, err := agent.Provider.Chat(
|
||||
|
||||
ctx,
|
||||
|
||||
[]providers.Message{{Role: "user", Content: mergePrompt}},
|
||||
|
||||
nil,
|
||||
|
||||
agent.Model,
|
||||
|
||||
map[string]any{
|
||||
"max_tokens": 1024,
|
||||
|
||||
"temperature": 0.3,
|
||||
|
||||
"prompt_cache_key": agent.ID,
|
||||
},
|
||||
)
|
||||
|
||||
if err == nil {
|
||||
finalSummary = resp.Content
|
||||
} else {
|
||||
finalSummary = s1 + " " + s2
|
||||
}
|
||||
} else {
|
||||
finalSummary, _ = al.summarizeBatch(ctx, agent, validMessages, summary)
|
||||
}
|
||||
|
||||
if omitted && finalSummary != "" {
|
||||
finalSummary += "\n[Note: Some oversized messages were omitted from this summary for efficiency.]"
|
||||
}
|
||||
|
||||
if finalSummary != "" {
|
||||
if err := agent.Sessions.CompactOldTurns(sessionKey, 4, finalSummary); err != nil {
|
||||
logger.ErrorCF("agent", "CompactOldTurns failed, falling back",
|
||||
|
||||
map[string]any{"error": err.Error()})
|
||||
|
||||
agent.Sessions.SetSummary(sessionKey, finalSummary)
|
||||
|
||||
agent.Sessions.TruncateHistory(sessionKey, 4)
|
||||
|
||||
agent.Sessions.Save(sessionKey)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// summarizeBatch summarizes a batch of messages.
|
||||
|
||||
func (al *AgentLoop) summarizeBatch(
|
||||
ctx context.Context,
|
||||
|
||||
agent *AgentInstance,
|
||||
|
||||
batch []providers.Message,
|
||||
|
||||
existingSummary string,
|
||||
) (string, error) {
|
||||
var sb strings.Builder
|
||||
|
||||
sb.WriteString("Provide a concise summary of this conversation segment, preserving core context and key points.\n")
|
||||
|
||||
if agent.ContextBuilder.HasActivePlan() {
|
||||
sb.WriteString("Note: Active plan in MEMORY.md. Preserve plan progress references.\n")
|
||||
}
|
||||
|
||||
if existingSummary != "" {
|
||||
sb.WriteString("Existing context: ")
|
||||
|
||||
sb.WriteString(existingSummary)
|
||||
|
||||
sb.WriteString("\n")
|
||||
}
|
||||
|
||||
sb.WriteString("\nCONVERSATION:\n")
|
||||
|
||||
for _, m := range batch {
|
||||
fmt.Fprintf(&sb, "%s: %s\n", m.Role, m.Content)
|
||||
}
|
||||
|
||||
prompt := sb.String()
|
||||
|
||||
response, err := agent.Provider.Chat(
|
||||
|
||||
ctx,
|
||||
|
||||
[]providers.Message{{Role: "user", Content: prompt}},
|
||||
|
||||
nil,
|
||||
|
||||
agent.Model,
|
||||
|
||||
map[string]any{
|
||||
"max_tokens": 1024,
|
||||
|
||||
"temperature": 0.3,
|
||||
|
||||
"prompt_cache_key": agent.ID,
|
||||
},
|
||||
)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
return response.Content, nil
|
||||
}
|
||||
|
||||
// estimateTokens estimates the number of tokens in a message list.
|
||||
|
||||
// Uses a safe heuristic of 2.5 characters per token to account for CJK and other
|
||||
|
||||
// overheads better than the previous 3 chars/token.
|
||||
|
||||
func (al *AgentLoop) estimateTokens(messages []providers.Message) int {
|
||||
totalChars := 0
|
||||
|
||||
for _, m := range messages {
|
||||
totalChars += utf8.RuneCountInString(m.Content)
|
||||
}
|
||||
|
||||
// 2.5 chars per token = totalChars * 2 / 5
|
||||
|
||||
return totalChars * 2 / 5
|
||||
}
|
||||
317
pkg/agent/loop_streaming.go
Normal file
317
pkg/agent/loop_streaming.go
Normal file
|
|
@ -0,0 +1,317 @@
|
|||
package agent
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"strings"
|
||||
"time"
|
||||
"unicode/utf8"
|
||||
|
||||
"github.com/sipeed/picoclaw/pkg/bus"
|
||||
"github.com/sipeed/picoclaw/pkg/logger"
|
||||
"github.com/sipeed/picoclaw/pkg/providers"
|
||||
"github.com/sipeed/picoclaw/pkg/providers/protocoltypes"
|
||||
"github.com/sipeed/picoclaw/pkg/utils"
|
||||
)
|
||||
|
||||
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(),
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// streamingReasoningLines is the number of lines reserved for reasoning
|
||||
|
||||
// in the streaming display. The remaining lines go to content.
|
||||
|
||||
const streamingReasoningLines = 6
|
||||
|
||||
// buildStreamingDisplay builds a fixed-height status bubble for streaming.
|
||||
|
||||
//
|
||||
|
||||
// Layout when reasoning is active (reasoning only or both):
|
||||
|
||||
//
|
||||
|
||||
// 🧠 Thinking...
|
||||
|
||||
// ━━━━━━━━━━
|
||||
|
||||
// <reasoning tail — streamingReasoningLines lines>
|
||||
|
||||
// ━━━━━━━━━━
|
||||
|
||||
// <content tail — remaining lines> (or blank if content is empty)
|
||||
|
||||
// █
|
||||
|
||||
//
|
||||
|
||||
// Layout when no reasoning (content only):
|
||||
|
||||
//
|
||||
|
||||
// <content tail — streamingDisplayLines lines>
|
||||
|
||||
// █
|
||||
|
||||
func buildStreamingDisplay(content, reasoning string) string {
|
||||
if reasoning == "" {
|
||||
// No reasoning — full window for content.
|
||||
|
||||
return utils.TailPad(content, streamingDisplayLines, maxEntryLineWidth) + " \u2589"
|
||||
}
|
||||
|
||||
var sb strings.Builder
|
||||
|
||||
// Header
|
||||
|
||||
if content == "" {
|
||||
sb.WriteString("\U0001f9e0 Thinking...\n")
|
||||
} else {
|
||||
sb.WriteString("\U0001f9e0 Thought, now responding...\n")
|
||||
}
|
||||
|
||||
sb.WriteString(statusSeparator)
|
||||
|
||||
// Reasoning window
|
||||
|
||||
headerLines := 2 // header + separator
|
||||
|
||||
footerLines := 1 // separator before content
|
||||
|
||||
contentLines := streamingDisplayLines - headerLines - footerLines - streamingReasoningLines
|
||||
|
||||
if contentLines < 3 {
|
||||
contentLines = 3
|
||||
}
|
||||
|
||||
rLines := streamingDisplayLines - headerLines - footerLines - contentLines
|
||||
|
||||
sb.WriteString(utils.TailPad(reasoning, rLines, maxEntryLineWidth))
|
||||
|
||||
sb.WriteByte('\n')
|
||||
|
||||
sb.WriteString(statusSeparator)
|
||||
|
||||
// Content window (may be blank padding if content hasn't started)
|
||||
|
||||
sb.WriteString(utils.TailPad(content, contentLines, maxEntryLineWidth))
|
||||
|
||||
sb.WriteString(" \u2589")
|
||||
|
||||
return sb.String()
|
||||
}
|
||||
|
||||
// consumeStreamWithRepetitionDetection reads StreamEvents from ch, accumulates
|
||||
|
||||
// content and tool calls, and runs repetition detection every checkInterval runes.
|
||||
|
||||
// If repetition is detected, cancelFn is called to abort the HTTP request and
|
||||
|
||||
// the function returns the partial response with detected=true.
|
||||
|
||||
func consumeStreamWithRepetitionDetection(
|
||||
ch <-chan protocoltypes.StreamEvent,
|
||||
|
||||
cancelFn context.CancelFunc,
|
||||
|
||||
checkInterval int,
|
||||
|
||||
onChunk func(content, reasoning string),
|
||||
) (*providers.LLMResponse, bool, error) {
|
||||
var content strings.Builder
|
||||
|
||||
var reasoning strings.Builder
|
||||
|
||||
var toolCalls []streamToolCallAcc
|
||||
|
||||
var finishReason string
|
||||
|
||||
var usage *providers.UsageInfo
|
||||
|
||||
runesSinceLastCheck := 0
|
||||
|
||||
for ev := range ch {
|
||||
if ev.Err != nil {
|
||||
return nil, false, ev.Err
|
||||
}
|
||||
|
||||
updated := false
|
||||
|
||||
if ev.ContentDelta != "" {
|
||||
content.WriteString(ev.ContentDelta)
|
||||
|
||||
runesSinceLastCheck += utf8.RuneCountInString(ev.ContentDelta)
|
||||
|
||||
updated = true
|
||||
}
|
||||
|
||||
if ev.ReasoningDelta != "" {
|
||||
reasoning.WriteString(ev.ReasoningDelta)
|
||||
|
||||
updated = true
|
||||
}
|
||||
|
||||
if updated && onChunk != nil {
|
||||
onChunk(content.String(), reasoning.String())
|
||||
}
|
||||
|
||||
if ev.FinishReason != "" {
|
||||
finishReason = ev.FinishReason
|
||||
}
|
||||
|
||||
if ev.Usage != nil {
|
||||
usage = ev.Usage
|
||||
}
|
||||
|
||||
for _, tc := range ev.ToolCallDeltas {
|
||||
for len(toolCalls) <= tc.Index {
|
||||
toolCalls = append(toolCalls, streamToolCallAcc{})
|
||||
}
|
||||
|
||||
if tc.ID != "" {
|
||||
toolCalls[tc.Index].id = tc.ID
|
||||
}
|
||||
|
||||
if tc.Name != "" {
|
||||
toolCalls[tc.Index].name = tc.Name
|
||||
}
|
||||
|
||||
toolCalls[tc.Index].args.WriteString(tc.ArgumentsDelta)
|
||||
}
|
||||
|
||||
// Run repetition detection periodically on accumulated content.
|
||||
|
||||
if runesSinceLastCheck >= checkInterval && content.Len() > 2000 {
|
||||
runesSinceLastCheck = 0
|
||||
|
||||
if utils.DetectRepetitionLoop(content.String()) {
|
||||
cancelFn()
|
||||
|
||||
// Drain remaining events so the producer goroutine can exit.
|
||||
|
||||
for range ch {
|
||||
}
|
||||
|
||||
resp := buildAccumulatedResponse(content.String(), reasoning.String(), toolCalls, finishReason, usage)
|
||||
|
||||
return resp, true, nil
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
resp := buildAccumulatedResponse(content.String(), reasoning.String(), toolCalls, finishReason, usage)
|
||||
|
||||
return resp, false, nil
|
||||
}
|
||||
|
||||
// streamToolCallAcc accumulates streamed tool call fragments.
|
||||
|
||||
type streamToolCallAcc struct {
|
||||
id string
|
||||
|
||||
name string
|
||||
|
||||
args strings.Builder
|
||||
}
|
||||
|
||||
// buildAccumulatedResponse constructs an LLMResponse from accumulated stream data.
|
||||
|
||||
func buildAccumulatedResponse(
|
||||
content, reasoning string,
|
||||
|
||||
toolCalls []streamToolCallAcc,
|
||||
|
||||
finishReason string,
|
||||
|
||||
usage *providers.UsageInfo,
|
||||
) *providers.LLMResponse {
|
||||
resp := &providers.LLMResponse{
|
||||
Content: content,
|
||||
|
||||
Reasoning: reasoning,
|
||||
|
||||
FinishReason: finishReason,
|
||||
|
||||
Usage: usage,
|
||||
}
|
||||
|
||||
for _, tc := range toolCalls {
|
||||
arguments := make(map[string]any)
|
||||
|
||||
argStr := tc.args.String()
|
||||
|
||||
if argStr != "" {
|
||||
if err := json.Unmarshal([]byte(argStr), &arguments); err != nil {
|
||||
arguments["raw"] = argStr
|
||||
}
|
||||
}
|
||||
|
||||
resp.ToolCalls = append(resp.ToolCalls, providers.ToolCall{
|
||||
ID: tc.id,
|
||||
|
||||
Name: tc.name,
|
||||
|
||||
Arguments: arguments,
|
||||
})
|
||||
}
|
||||
|
||||
return resp
|
||||
}
|
||||
701
pkg/agent/loop_task.go
Normal file
701
pkg/agent/loop_task.go
Normal file
|
|
@ -0,0 +1,701 @@
|
|||
package agent
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
"unicode"
|
||||
"unicode/utf8"
|
||||
|
||||
"github.com/sipeed/picoclaw/pkg/providers"
|
||||
"github.com/sipeed/picoclaw/pkg/utils"
|
||||
)
|
||||
|
||||
// activeTask tracks a running agent task for live status and intervention.
|
||||
|
||||
type activeTask struct {
|
||||
Description string
|
||||
|
||||
Result string // LLM response summary for completion notification
|
||||
|
||||
Iteration int
|
||||
|
||||
MaxIter int
|
||||
|
||||
StartedAt time.Time
|
||||
|
||||
cancel context.CancelFunc
|
||||
|
||||
interrupt chan string // buffered 1, for user message injection
|
||||
|
||||
toolLog []toolLogEntry
|
||||
|
||||
lastError *toolLogEntry // sticky: most recent error, persists across iterations
|
||||
|
||||
projectDir string // detected from exec cd target (authoritative)
|
||||
|
||||
fileCommonDir string // LCP of file paths relative to workspace (fallback)
|
||||
|
||||
streamedChunks bool // true after onChunk fires at least once
|
||||
|
||||
messageContent string // last content sent by the message tool (for inclusion in completion)
|
||||
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
// toolLogEntry records a single tool call for the live terminal view.
|
||||
|
||||
type toolLogEntry struct {
|
||||
Name string
|
||||
|
||||
ArgsSnip string // first ~80 chars of args
|
||||
|
||||
Result string // "✓ 4.9s" or "✗ 3.2s"
|
||||
|
||||
ErrDetail string // non-empty on error — e.g. "Exit code: exit status 1"
|
||||
}
|
||||
|
||||
// maxToolLogEntries limits the sliding window of tool log entries
|
||||
|
||||
// kept in memory and displayed in status messages.
|
||||
|
||||
const maxToolLogEntries = 5
|
||||
|
||||
// Task reminder constants and helpers.
|
||||
|
||||
const (
|
||||
taskReminderMaxChars = 500
|
||||
|
||||
blockerMaxChars = 200
|
||||
)
|
||||
|
||||
func shouldInjectReminder(iteration, interval int) bool {
|
||||
if interval <= 0 {
|
||||
return false
|
||||
}
|
||||
|
||||
return iteration > 1 && iteration%interval == 0
|
||||
}
|
||||
|
||||
func buildTaskReminder(userMessage string, lastBlocker string) providers.Message {
|
||||
truncatedTask := utils.Truncate(userMessage, taskReminderMaxChars)
|
||||
|
||||
var content string
|
||||
|
||||
if lastBlocker != "" {
|
||||
truncatedBlocker := utils.Truncate(lastBlocker, blockerMaxChars)
|
||||
|
||||
content = fmt.Sprintf(
|
||||
"[TASK REMINDER]\nOriginal task:\n---\n%s\n---\nLast blocker:\n---\n%s\n---\nFix the blocker if essential, or find an alternative. If all steps are complete, move on.",
|
||||
truncatedTask,
|
||||
truncatedBlocker,
|
||||
)
|
||||
} else {
|
||||
content = fmt.Sprintf(
|
||||
"[TASK REMINDER]\nOriginal task:\n---\n%s\n---\nIf all steps of the original task are complete, move on. Otherwise, continue with the next step.",
|
||||
truncatedTask,
|
||||
)
|
||||
}
|
||||
|
||||
return providers.Message{
|
||||
Role: "user",
|
||||
|
||||
Content: content,
|
||||
}
|
||||
}
|
||||
|
||||
// cdPrefixPattern matches "cd /some/path && " at the start of a shell command.
|
||||
|
||||
// Group 1 captures the target directory path.
|
||||
|
||||
var cdPrefixPattern = regexp.MustCompile(`^cd\s+(\S+)\s*&&\s*`)
|
||||
|
||||
// optFlagPattern matches option flags like --verbose, -v, --timeout=60, -q.
|
||||
|
||||
// Only standalone flags are removed; flags whose value is the next positional
|
||||
|
||||
// argument (e.g. "-A 20") are kept because removing them would lose context.
|
||||
|
||||
var optFlagPattern = regexp.MustCompile(`\s+--?\w[\w-]*(=\S*)?`)
|
||||
|
||||
// extractExecProjectDir extracts the basename of an exec cd target.
|
||||
|
||||
// Returns "" if the command has no cd prefix.
|
||||
|
||||
func extractExecProjectDir(args map[string]any) string {
|
||||
cmd, _ := args["command"].(string)
|
||||
|
||||
if cmd == "" {
|
||||
return ""
|
||||
}
|
||||
|
||||
m := cdPrefixPattern.FindStringSubmatch(cmd)
|
||||
|
||||
if len(m) < 2 {
|
||||
return ""
|
||||
}
|
||||
|
||||
cdPath := strings.TrimRight(m[1], "/\\")
|
||||
|
||||
if idx := strings.LastIndex(cdPath, "/"); idx >= 0 {
|
||||
return cdPath[idx+1:]
|
||||
}
|
||||
|
||||
if idx := strings.LastIndex(cdPath, "\\"); idx >= 0 {
|
||||
return cdPath[idx+1:]
|
||||
}
|
||||
|
||||
return cdPath
|
||||
}
|
||||
|
||||
// fileParentRelDir returns the parent directory of a file path, relative to
|
||||
|
||||
// workspace. Returns "" if the path is not under workspace or has no parent.
|
||||
|
||||
func fileParentRelDir(filePath, workspace string) string {
|
||||
ws := strings.TrimRight(workspace, "/\\")
|
||||
|
||||
if ws == "" {
|
||||
return ""
|
||||
}
|
||||
|
||||
rest := strings.TrimPrefix(filePath, ws)
|
||||
|
||||
if rest == filePath {
|
||||
return "" // not under workspace
|
||||
}
|
||||
|
||||
rest = strings.TrimLeft(rest, "/\\")
|
||||
|
||||
// Remove the filename — keep only the directory part
|
||||
|
||||
if idx := strings.LastIndexAny(rest, "/\\"); idx >= 0 {
|
||||
return rest[:idx]
|
||||
}
|
||||
|
||||
return "" // file is directly under workspace, no meaningful dir
|
||||
}
|
||||
|
||||
// commonDirPrefix computes the longest common directory prefix of two
|
||||
|
||||
// slash-separated paths. Returns "" if there is no common component.
|
||||
|
||||
func commonDirPrefix(a, b string) string {
|
||||
partsA := strings.Split(a, "/")
|
||||
|
||||
partsB := strings.Split(b, "/")
|
||||
|
||||
n := len(partsA)
|
||||
|
||||
if len(partsB) < n {
|
||||
n = len(partsB)
|
||||
}
|
||||
|
||||
common := 0
|
||||
|
||||
for i := 0; i < n; i++ {
|
||||
if partsA[i] != partsB[i] {
|
||||
break
|
||||
}
|
||||
|
||||
common = i + 1
|
||||
}
|
||||
|
||||
if common == 0 {
|
||||
return ""
|
||||
}
|
||||
|
||||
return strings.Join(partsA[:common], "/")
|
||||
}
|
||||
|
||||
// displayProjectDir returns the project directory name for status display.
|
||||
|
||||
// Prefers the authoritative exec-based projectDir; falls back to the
|
||||
|
||||
// basename of the file-based common directory.
|
||||
|
||||
func displayProjectDir(task *activeTask) string {
|
||||
if task.projectDir != "" {
|
||||
return task.projectDir
|
||||
}
|
||||
|
||||
if task.fileCommonDir != "" {
|
||||
dir := task.fileCommonDir
|
||||
|
||||
if idx := strings.LastIndex(dir, "/"); idx >= 0 {
|
||||
return dir[idx+1:]
|
||||
}
|
||||
|
||||
return dir
|
||||
}
|
||||
|
||||
return ""
|
||||
}
|
||||
|
||||
// buildArgsSnippet produces a human-friendly snippet for the tool log.
|
||||
|
||||
// For exec: extracts the command and strips the leading "cd <workspace> && ".
|
||||
|
||||
// For file tools: extracts the path and strips the workspace prefix.
|
||||
|
||||
// Falls back to raw JSON truncation.
|
||||
|
||||
func buildArgsSnippet(toolName string, args map[string]any, workspace string) string {
|
||||
switch toolName {
|
||||
case "exec":
|
||||
|
||||
cmd, _ := args["command"].(string)
|
||||
|
||||
if cmd == "" {
|
||||
break
|
||||
}
|
||||
|
||||
cmd = cdPrefixPattern.ReplaceAllString(cmd, "")
|
||||
|
||||
cmd = optFlagPattern.ReplaceAllString(cmd, "")
|
||||
|
||||
return utils.Truncate(cmd, 80)
|
||||
|
||||
case "read_file", "write_file", "edit_file", "append_file", "list_dir":
|
||||
|
||||
path, _ := args["path"].(string)
|
||||
|
||||
if path == "" {
|
||||
break
|
||||
}
|
||||
|
||||
if workspace != "" {
|
||||
path = strings.TrimPrefix(path, workspace)
|
||||
|
||||
path = strings.TrimPrefix(path, "/")
|
||||
}
|
||||
|
||||
// Prioritize filename: if path is too long, show "…/filename"
|
||||
|
||||
const maxPath = 60
|
||||
|
||||
if runes := []rune(path); len(runes) > maxPath {
|
||||
// Find last slash to extract filename
|
||||
|
||||
if lastSlash := strings.LastIndex(path, "/"); lastSlash >= 0 {
|
||||
filename := path[lastSlash:] // includes "/"
|
||||
|
||||
dirBudget := maxPath - len([]rune(filename)) - 1 // 1 for "…"
|
||||
|
||||
if dirBudget > 0 {
|
||||
dir := []rune(path[:lastSlash])
|
||||
|
||||
if len(dir) > dirBudget {
|
||||
dir = dir[:dirBudget]
|
||||
}
|
||||
|
||||
path = string(dir) + "\u2026" + filename
|
||||
} else {
|
||||
path = "\u2026" + filename
|
||||
}
|
||||
} else {
|
||||
path = utils.Truncate(path, maxPath)
|
||||
}
|
||||
}
|
||||
|
||||
return path
|
||||
}
|
||||
|
||||
// Default: raw JSON truncated
|
||||
|
||||
argsJSON, _ := json.Marshal(args)
|
||||
|
||||
return utils.Truncate(string(argsJSON), 80)
|
||||
}
|
||||
|
||||
// maxEntryLineWidth is the max rune count for a single-line log entry.
|
||||
|
||||
// Telegram chat bubbles on mobile are roughly 40-45 chars wide.
|
||||
|
||||
const maxEntryLineWidth = 42
|
||||
|
||||
// isFileToolEntry returns true if the entry name contains a file-operation tool.
|
||||
|
||||
func isFileToolEntry(name string) bool {
|
||||
for _, t := range []string{"read_file", "write_file", "edit_file", "append_file", "list_dir"} {
|
||||
if strings.Contains(name, t) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
// formatCompactEntry formats a finished tool log entry as a fixed single line.
|
||||
|
||||
// The result marker (✓/✗) is always shown at the end regardless of truncation.
|
||||
|
||||
// File tools omit duration (always near-instant); paths truncate from the
|
||||
|
||||
// start so the filename is always visible.
|
||||
|
||||
func formatCompactEntry(entry toolLogEntry) string {
|
||||
result := entry.Result
|
||||
|
||||
if result == "" {
|
||||
result = "\u23F3" // ⏳
|
||||
}
|
||||
|
||||
// File tools: strip duration, keep only marker (✓/✗/⏳)
|
||||
|
||||
isFile := isFileToolEntry(entry.Name)
|
||||
|
||||
if isFile {
|
||||
if r := []rune(result); len(r) > 0 {
|
||||
result = string(r[0:1]) // just the symbol
|
||||
}
|
||||
}
|
||||
|
||||
// Budget for ArgsSnip: total - name - " " - " " - result
|
||||
|
||||
nameLen := utf8.RuneCountInString(entry.Name)
|
||||
|
||||
resultLen := utf8.RuneCountInString(result)
|
||||
|
||||
argsBudget := maxEntryLineWidth - nameLen - 1 - 1 - resultLen
|
||||
|
||||
args := entry.ArgsSnip
|
||||
|
||||
if args != "" && argsBudget > 3 {
|
||||
argsRunes := []rune(args)
|
||||
|
||||
if len(argsRunes) > argsBudget {
|
||||
// Paths: truncate from the start, keeping the filename visible
|
||||
|
||||
if strings.Contains(args, "/") {
|
||||
args = "\u2026" + string(argsRunes[len(argsRunes)-argsBudget+1:])
|
||||
} else {
|
||||
args = string(argsRunes[:argsBudget-1]) + "\u2026"
|
||||
}
|
||||
}
|
||||
|
||||
var sb strings.Builder
|
||||
|
||||
sb.Grow(len(entry.Name) + 1 + len(args) + 1 + len(result))
|
||||
|
||||
sb.WriteString(entry.Name)
|
||||
|
||||
sb.WriteByte(' ')
|
||||
|
||||
sb.WriteString(args)
|
||||
|
||||
sb.WriteByte(' ')
|
||||
|
||||
sb.WriteString(result)
|
||||
|
||||
return sb.String()
|
||||
}
|
||||
|
||||
// No room for args or args empty
|
||||
|
||||
var sb strings.Builder
|
||||
|
||||
sb.Grow(len(entry.Name) + 1 + len(result))
|
||||
|
||||
sb.WriteString(entry.Name)
|
||||
|
||||
sb.WriteByte(' ')
|
||||
|
||||
sb.WriteString(result)
|
||||
|
||||
return sb.String()
|
||||
}
|
||||
|
||||
// formatLatestEntry formats the latest entry command without its result marker.
|
||||
|
||||
// Since the result goes on the next line, the full width is available for the command.
|
||||
|
||||
func formatLatestEntry(entry toolLogEntry) string {
|
||||
nameLen := utf8.RuneCountInString(entry.Name)
|
||||
|
||||
argsBudget := maxEntryLineWidth - nameLen - 1 // name + space + args (no result)
|
||||
|
||||
args := entry.ArgsSnip
|
||||
|
||||
if args != "" && argsBudget > 3 {
|
||||
argsRunes := []rune(args)
|
||||
|
||||
if len(argsRunes) > argsBudget {
|
||||
if strings.Contains(args, "/") {
|
||||
args = "\u2026" + string(argsRunes[len(argsRunes)-argsBudget+1:])
|
||||
} else {
|
||||
args = string(argsRunes[:argsBudget-1]) + "\u2026"
|
||||
}
|
||||
}
|
||||
|
||||
var sb strings.Builder
|
||||
|
||||
sb.Grow(len(entry.Name) + 1 + len(args))
|
||||
|
||||
sb.WriteString(entry.Name)
|
||||
|
||||
sb.WriteByte(' ')
|
||||
|
||||
sb.WriteString(args)
|
||||
|
||||
return sb.String()
|
||||
}
|
||||
|
||||
return entry.Name
|
||||
}
|
||||
|
||||
// compressRepeats reduces runs of 3+ identical non-alphanumeric, non-space
|
||||
|
||||
// characters to just 2. e.g. "======" → "==", "---" → "--".
|
||||
|
||||
func compressRepeats(s string) string {
|
||||
runes := []rune(s)
|
||||
|
||||
if len(runes) < 3 {
|
||||
return s
|
||||
}
|
||||
|
||||
var sb strings.Builder
|
||||
|
||||
sb.Grow(len(s))
|
||||
|
||||
i := 0
|
||||
|
||||
for i < len(runes) {
|
||||
r := runes[i]
|
||||
|
||||
if !unicode.IsLetter(r) && !unicode.IsDigit(r) && !unicode.IsSpace(r) {
|
||||
j := i + 1
|
||||
|
||||
for j < len(runes) && runes[j] == r {
|
||||
j++
|
||||
}
|
||||
|
||||
if j-i >= 3 {
|
||||
sb.WriteRune(r)
|
||||
|
||||
sb.WriteRune(r)
|
||||
|
||||
i = j
|
||||
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
||||
sb.WriteRune(r)
|
||||
|
||||
i++
|
||||
}
|
||||
|
||||
return sb.String()
|
||||
}
|
||||
|
||||
// Display layout constants.
|
||||
|
||||
const (
|
||||
displayPastEntries = 4 // number of compact 1-line past entries
|
||||
|
||||
displayErrorLines = 5 // content lines inside the error code block
|
||||
|
||||
statusSeparator = "\u2501\u2501\u2501\u2501\u2501\u2501\u2501\u2501\u2501\u2501\n"
|
||||
|
||||
streamingDisplayLines = 17 // line count matching buildRichStatus output
|
||||
|
||||
)
|
||||
|
||||
// buildRichStatus builds a fixed-height terminal-like status display.
|
||||
|
||||
//
|
||||
|
||||
// Layout (always the same number of lines):
|
||||
|
||||
//
|
||||
|
||||
// 🔄 Task in progress (N/M) header
|
||||
|
||||
// 📁 workspace-path header
|
||||
|
||||
// ━━━━━━━━━━ separator
|
||||
|
||||
// [N] compact-past-1 ✓ Xs past (1 line each)
|
||||
|
||||
// [N] compact-past-2 ✗ Xs past
|
||||
|
||||
// [N] compact-past-3 ✓ Xs past
|
||||
|
||||
// [N] compact-past-4 ✓ Xs past
|
||||
|
||||
// [N] latest-command latest (no result, wider args)
|
||||
|
||||
// ⏳ latest result
|
||||
|
||||
// reserved
|
||||
|
||||
// ``` error fence
|
||||
|
||||
// err-line / placeholder error body (5 lines)
|
||||
|
||||
// ``` error fence
|
||||
|
||||
// ↩️ Reply to intervene footer (background only)
|
||||
|
||||
func buildRichStatus(task *activeTask, isBackground bool, workspace string) string {
|
||||
task.mu.Lock()
|
||||
|
||||
defer task.mu.Unlock()
|
||||
|
||||
var sb strings.Builder
|
||||
|
||||
// --- Header ---
|
||||
|
||||
sb.WriteString("\U0001F504 Task in progress (")
|
||||
|
||||
sb.WriteString(strconv.Itoa(task.Iteration))
|
||||
|
||||
sb.WriteByte('/')
|
||||
|
||||
sb.WriteString(strconv.Itoa(task.MaxIter))
|
||||
|
||||
sb.WriteString(")\n")
|
||||
|
||||
// Project directory: exec cd (authoritative) → file LCP → workspace basename
|
||||
|
||||
sb.WriteString("\U0001F4C1 ")
|
||||
|
||||
if dir := displayProjectDir(task); dir != "" {
|
||||
sb.WriteString(dir)
|
||||
} else if workspace != "" {
|
||||
project := strings.TrimRight(workspace, "/\\")
|
||||
|
||||
if idx := strings.LastIndex(project, "/"); idx >= 0 {
|
||||
project = project[idx+1:]
|
||||
} else if idx := strings.LastIndex(project, "\\"); idx >= 0 {
|
||||
project = project[idx+1:]
|
||||
}
|
||||
|
||||
sb.WriteString(project)
|
||||
}
|
||||
|
||||
sb.WriteByte('\n')
|
||||
|
||||
sb.WriteString(statusSeparator)
|
||||
|
||||
// --- Task entries (displayPastEntries + 2 lines for latest) ---
|
||||
|
||||
entries := task.toolLog
|
||||
|
||||
if len(entries) > maxToolLogEntries {
|
||||
entries = entries[len(entries)-maxToolLogEntries:]
|
||||
}
|
||||
|
||||
var pastEntries []toolLogEntry
|
||||
|
||||
var latest *toolLogEntry
|
||||
|
||||
if len(entries) > 0 {
|
||||
latest = &entries[len(entries)-1]
|
||||
|
||||
if len(entries) > 1 {
|
||||
start := len(entries) - 1 - displayPastEntries
|
||||
|
||||
if start < 0 {
|
||||
start = 0
|
||||
}
|
||||
|
||||
pastEntries = entries[start : len(entries)-1]
|
||||
}
|
||||
}
|
||||
|
||||
// Past entries: exactly displayPastEntries lines (pad if fewer)
|
||||
|
||||
for i := 0; i < displayPastEntries; i++ {
|
||||
if i < len(pastEntries) {
|
||||
sb.WriteString(formatCompactEntry(pastEntries[i]))
|
||||
} else {
|
||||
sb.WriteString("\u2800")
|
||||
}
|
||||
|
||||
sb.WriteByte('\n')
|
||||
}
|
||||
|
||||
// Latest entry: command on one line, result on next
|
||||
|
||||
if latest != nil {
|
||||
sb.WriteString(formatLatestEntry(*latest))
|
||||
|
||||
sb.WriteByte('\n')
|
||||
|
||||
sb.WriteString(" ")
|
||||
|
||||
if latest.Result != "" {
|
||||
sb.WriteString(latest.Result)
|
||||
} else {
|
||||
sb.WriteString("\u23F3")
|
||||
}
|
||||
|
||||
sb.WriteByte('\n')
|
||||
} else {
|
||||
sb.WriteString("\u23F3 waiting...\n")
|
||||
|
||||
sb.WriteString("\u2800\n")
|
||||
}
|
||||
|
||||
// Reserved (1 line)
|
||||
|
||||
sb.WriteString("\u2800\n")
|
||||
|
||||
// --- Error region (code fence, no separator) ---
|
||||
|
||||
sb.WriteString("```\n")
|
||||
|
||||
errEntry := task.lastError
|
||||
|
||||
if errEntry != nil {
|
||||
sb.WriteString("\u274C ")
|
||||
|
||||
sb.WriteString(formatCompactEntry(*errEntry))
|
||||
|
||||
sb.WriteByte('\n')
|
||||
|
||||
var detailLines []string
|
||||
|
||||
if errEntry.ErrDetail != "" {
|
||||
detailLines = strings.Split(errEntry.ErrDetail, "\n")
|
||||
}
|
||||
|
||||
for i := 0; i < displayErrorLines-1; i++ {
|
||||
if i < len(detailLines) {
|
||||
line := compressRepeats(detailLines[i])
|
||||
|
||||
if runes := []rune(line); len(runes) > maxEntryLineWidth {
|
||||
line = string(runes[:maxEntryLineWidth-1]) + "\u2026"
|
||||
}
|
||||
|
||||
sb.WriteString(line)
|
||||
} else {
|
||||
sb.WriteString("\u2800")
|
||||
}
|
||||
|
||||
sb.WriteByte('\n')
|
||||
}
|
||||
} else {
|
||||
sb.WriteString("\u2714 No errors\n")
|
||||
|
||||
for i := 0; i < displayErrorLines-1; i++ {
|
||||
sb.WriteString("\u2800\n")
|
||||
}
|
||||
}
|
||||
|
||||
sb.WriteString("```\n")
|
||||
|
||||
if isBackground {
|
||||
sb.WriteString("\u21A9\uFE0F Reply to intervene")
|
||||
}
|
||||
|
||||
return sb.String()
|
||||
}
|
||||
|
|
@ -4,6 +4,7 @@ import (
|
|||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
|
||||
"github.com/caarlos0/env/v11"
|
||||
|
|
@ -140,16 +141,6 @@ type AgentConfig struct {
|
|||
Subagents *SubagentsConfig `json:"subagents,omitempty"`
|
||||
}
|
||||
|
||||
type SubagentsConfig struct {
|
||||
Enabled bool `json:"enabled,omitempty"`
|
||||
AllowAgents []string `json:"allow_agents,omitempty"`
|
||||
Model *AgentModelConfig `json:"model,omitempty"`
|
||||
}
|
||||
|
||||
type PeerMatch struct {
|
||||
Kind string `json:"kind"`
|
||||
ID string `json:"id"`
|
||||
}
|
||||
|
||||
type BindingMatch struct {
|
||||
Channel string `json:"channel"`
|
||||
|
|
|
|||
14
pkg/config/config_ext.go
Normal file
14
pkg/config/config_ext.go
Normal file
|
|
@ -0,0 +1,14 @@
|
|||
package config
|
||||
|
||||
// SubagentsConfig holds fork-specific subagent orchestration settings.
|
||||
type SubagentsConfig struct {
|
||||
Enabled bool `json:"enabled,omitempty"`
|
||||
AllowAgents []string `json:"allow_agents,omitempty"`
|
||||
Model *AgentModelConfig `json:"model,omitempty"`
|
||||
}
|
||||
|
||||
// PeerMatch identifies a peer by kind (direct/group) and ID.
|
||||
type PeerMatch struct {
|
||||
Kind string `json:"kind"`
|
||||
ID string `json:"id"`
|
||||
}
|
||||
|
|
@ -10,84 +10,6 @@ type Tool interface {
|
|||
Execute(ctx context.Context, args map[string]any) *ToolResult
|
||||
}
|
||||
|
||||
// --- Request-scoped tool context (channel / chatID) ---
|
||||
//
|
||||
// Carried via context.Value so that concurrent tool calls each receive
|
||||
// their own immutable copy — no mutable state on singleton tool instances.
|
||||
//
|
||||
// Keys are unexported pointer-typed vars — guaranteed collision-free,
|
||||
// and only accessible through the helper functions below.
|
||||
|
||||
type toolCtxKey struct{ name string }
|
||||
|
||||
var (
|
||||
ctxKeyChannel = &toolCtxKey{"channel"}
|
||||
ctxKeyChatID = &toolCtxKey{"chatID"}
|
||||
)
|
||||
|
||||
// WithToolContext returns a child context carrying channel and chatID.
|
||||
func WithToolContext(ctx context.Context, channel, chatID string) context.Context {
|
||||
ctx = context.WithValue(ctx, ctxKeyChannel, channel)
|
||||
ctx = context.WithValue(ctx, ctxKeyChatID, chatID)
|
||||
return ctx
|
||||
}
|
||||
|
||||
// ToolChannel extracts the channel from ctx, or "" if unset.
|
||||
func ToolChannel(ctx context.Context) string {
|
||||
v, _ := ctx.Value(ctxKeyChannel).(string)
|
||||
return v
|
||||
}
|
||||
|
||||
// ToolChatID extracts the chatID from ctx, or "" if unset.
|
||||
func ToolChatID(ctx context.Context) string {
|
||||
v, _ := ctx.Value(ctxKeyChatID).(string)
|
||||
return v
|
||||
}
|
||||
|
||||
// AsyncCallback is a function type that async tools use to notify completion.
|
||||
// When an async tool finishes its work, it calls this callback with the result.
|
||||
//
|
||||
// The ctx parameter allows the callback to be canceled if the agent is shutting down.
|
||||
// The result parameter contains the tool's execution result.
|
||||
type AsyncCallback func(ctx context.Context, result *ToolResult)
|
||||
|
||||
// AsyncExecutor is an optional interface that tools can implement to support
|
||||
// asynchronous execution with completion callbacks.
|
||||
//
|
||||
// Unlike the old AsyncTool pattern (SetCallback + Execute), AsyncExecutor
|
||||
// receives the callback as a parameter of ExecuteAsync. This eliminates the
|
||||
// data race where concurrent calls could overwrite each other's callbacks
|
||||
// on a shared tool instance.
|
||||
//
|
||||
// This is useful for:
|
||||
// - Long-running operations that shouldn't block the agent loop
|
||||
// - Subagent spawns that complete independently
|
||||
// - Background tasks that need to report results later
|
||||
//
|
||||
// Example:
|
||||
//
|
||||
// func (t *SpawnTool) ExecuteAsync(ctx context.Context, args map[string]any, cb AsyncCallback) *ToolResult {
|
||||
// go func() {
|
||||
// result := t.runSubagent(ctx, args)
|
||||
// if cb != nil { cb(ctx, result) }
|
||||
// }()
|
||||
// return AsyncResult("Subagent spawned, will report back")
|
||||
// }
|
||||
type AsyncExecutor interface {
|
||||
Tool
|
||||
// ExecuteAsync runs the tool asynchronously. The callback cb will be
|
||||
// invoked (possibly from another goroutine) when the async operation
|
||||
// completes. cb is guaranteed to be non-nil by the caller (registry).
|
||||
ExecuteAsync(ctx context.Context, args map[string]any, cb AsyncCallback) *ToolResult
|
||||
}
|
||||
|
||||
// StatusProvider is an optional interface that tools can implement
|
||||
// to inject runtime status information into the system prompt.
|
||||
// Return an empty string to inject nothing.
|
||||
type StatusProvider interface {
|
||||
RuntimeStatus() string
|
||||
}
|
||||
|
||||
func ToolToSchema(tool Tool) map[string]any {
|
||||
return map[string]any{
|
||||
"type": "function",
|
||||
|
|
|
|||
81
pkg/tools/base_ext.go
Normal file
81
pkg/tools/base_ext.go
Normal file
|
|
@ -0,0 +1,81 @@
|
|||
package tools
|
||||
|
||||
import "context"
|
||||
|
||||
// --- Request-scoped tool context (channel / chatID) ---
|
||||
//
|
||||
// Carried via context.Value so that concurrent tool calls each receive
|
||||
// their own immutable copy — no mutable state on singleton tool instances.
|
||||
//
|
||||
// Keys are unexported pointer-typed vars — guaranteed collision-free,
|
||||
// and only accessible through the helper functions below.
|
||||
|
||||
type toolCtxKey struct{ name string }
|
||||
|
||||
var (
|
||||
ctxKeyChannel = &toolCtxKey{"channel"}
|
||||
ctxKeyChatID = &toolCtxKey{"chatID"}
|
||||
)
|
||||
|
||||
// WithToolContext returns a child context carrying channel and chatID.
|
||||
func WithToolContext(ctx context.Context, channel, chatID string) context.Context {
|
||||
ctx = context.WithValue(ctx, ctxKeyChannel, channel)
|
||||
ctx = context.WithValue(ctx, ctxKeyChatID, chatID)
|
||||
return ctx
|
||||
}
|
||||
|
||||
// ToolChannel extracts the channel from ctx, or "" if unset.
|
||||
func ToolChannel(ctx context.Context) string {
|
||||
v, _ := ctx.Value(ctxKeyChannel).(string)
|
||||
return v
|
||||
}
|
||||
|
||||
// ToolChatID extracts the chatID from ctx, or "" if unset.
|
||||
func ToolChatID(ctx context.Context) string {
|
||||
v, _ := ctx.Value(ctxKeyChatID).(string)
|
||||
return v
|
||||
}
|
||||
|
||||
// AsyncCallback is a function type that async tools use to notify completion.
|
||||
// When an async tool finishes its work, it calls this callback with the result.
|
||||
//
|
||||
// The ctx parameter allows the callback to be canceled if the agent is shutting down.
|
||||
// The result parameter contains the tool's execution result.
|
||||
type AsyncCallback func(ctx context.Context, result *ToolResult)
|
||||
|
||||
// AsyncExecutor is an optional interface that tools can implement to support
|
||||
// asynchronous execution with completion callbacks.
|
||||
//
|
||||
// Unlike the old AsyncTool pattern (SetCallback + Execute), AsyncExecutor
|
||||
// receives the callback as a parameter of ExecuteAsync. This eliminates the
|
||||
// data race where concurrent calls could overwrite each other's callbacks
|
||||
// on a shared tool instance.
|
||||
//
|
||||
// This is useful for:
|
||||
// - Long-running operations that shouldn't block the agent loop
|
||||
// - Subagent spawns that complete independently
|
||||
// - Background tasks that need to report results later
|
||||
//
|
||||
// Example:
|
||||
//
|
||||
// func (t *SpawnTool) ExecuteAsync(ctx context.Context, args map[string]any, cb AsyncCallback) *ToolResult {
|
||||
// go func() {
|
||||
// result := t.runSubagent(ctx, args)
|
||||
// if cb != nil { cb(ctx, result) }
|
||||
// }()
|
||||
// return AsyncResult("Subagent spawned, will report back")
|
||||
// }
|
||||
type AsyncExecutor interface {
|
||||
Tool
|
||||
// ExecuteAsync runs the tool asynchronously. The callback cb will be
|
||||
// invoked (possibly from another goroutine) when the async operation
|
||||
// completes. cb is guaranteed to be non-nil by the caller (registry).
|
||||
ExecuteAsync(ctx context.Context, args map[string]any, cb AsyncCallback) *ToolResult
|
||||
}
|
||||
|
||||
// StatusProvider is an optional interface that tools can implement
|
||||
// to inject runtime status information into the system prompt.
|
||||
// Return an empty string to inject nothing.
|
||||
type StatusProvider interface {
|
||||
RuntimeStatus() string
|
||||
}
|
||||
|
|
@ -2,15 +2,9 @@ package tools
|
|||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/sipeed/picoclaw/pkg/logger"
|
||||
"github.com/sipeed/picoclaw/pkg/providers"
|
||||
)
|
||||
|
||||
// NormalizeToolName keeps only lowercase ASCII letters.
|
||||
|
|
@ -79,107 +73,6 @@ func (r *ToolRegistry) Execute(ctx context.Context, name string, args map[string
|
|||
return r.ExecuteWithContext(ctx, name, args, "", "", nil)
|
||||
}
|
||||
|
||||
// ExecuteWithContext executes a tool with channel/chatID context and optional async callback.
|
||||
|
||||
// If the tool implements AsyncTool and a non-nil callback is provided,
|
||||
|
||||
// the callback will be set on the tool before execution.
|
||||
|
||||
func (r *ToolRegistry) ExecuteWithContext(
|
||||
ctx context.Context,
|
||||
|
||||
name string,
|
||||
|
||||
args map[string]any,
|
||||
|
||||
channel, chatID string,
|
||||
|
||||
asyncCallback AsyncCallback,
|
||||
) *ToolResult {
|
||||
logger.InfoCF("tool", "Tool execution started",
|
||||
|
||||
map[string]any{
|
||||
"tool": name,
|
||||
|
||||
"args": args,
|
||||
})
|
||||
|
||||
tool, ok := r.Get(name)
|
||||
|
||||
if !ok {
|
||||
available := strings.Join(r.List(), ", ")
|
||||
|
||||
logger.ErrorCF("tool", "Tool not found",
|
||||
|
||||
map[string]any{
|
||||
"tool": name,
|
||||
})
|
||||
|
||||
return ErrorResult(fmt.Sprintf(
|
||||
|
||||
"tool %q not found. Available tools: %s", name, available,
|
||||
)).WithError(fmt.Errorf("tool not found"))
|
||||
}
|
||||
|
||||
// If tool implements ContextualTool, set context
|
||||
|
||||
if contextualTool, ok := tool.(ContextualTool); ok && channel != "" && chatID != "" {
|
||||
contextualTool.SetContext(channel, chatID)
|
||||
}
|
||||
|
||||
// If tool implements AsyncTool and callback is provided, set callback
|
||||
|
||||
if asyncTool, ok := tool.(AsyncTool); ok && asyncCallback != nil {
|
||||
asyncTool.SetCallback(asyncCallback)
|
||||
|
||||
logger.DebugCF("tool", "Async callback injected",
|
||||
|
||||
map[string]any{
|
||||
"tool": name,
|
||||
})
|
||||
}
|
||||
|
||||
start := time.Now()
|
||||
|
||||
result := tool.Execute(ctx, args)
|
||||
|
||||
duration := time.Since(start)
|
||||
|
||||
// Log based on result type
|
||||
|
||||
if result.IsError {
|
||||
logger.ErrorCF("tool", "Tool execution failed",
|
||||
|
||||
map[string]any{
|
||||
"tool": name,
|
||||
|
||||
"duration": duration.Milliseconds(),
|
||||
|
||||
"error": result.ForLLM,
|
||||
})
|
||||
} else if result.Async {
|
||||
logger.InfoCF("tool", "Tool started (async)",
|
||||
|
||||
map[string]any{
|
||||
"tool": name,
|
||||
|
||||
"duration": duration.Milliseconds(),
|
||||
})
|
||||
} else {
|
||||
logger.InfoCF("tool", "Tool execution completed",
|
||||
|
||||
map[string]any{
|
||||
"tool": name,
|
||||
|
||||
"duration_ms": duration.Milliseconds(),
|
||||
|
||||
"result_length": len(result.ForLLM),
|
||||
})
|
||||
}
|
||||
|
||||
return result
|
||||
}
|
||||
|
||||
// sortedToolNames returns tool names in sorted order for deterministic iteration.
|
||||
|
||||
// This is critical for KV cache stability: non-deterministic map iteration would
|
||||
|
|
@ -216,62 +109,6 @@ func (r *ToolRegistry) GetDefinitions() []map[string]any {
|
|||
return definitions
|
||||
}
|
||||
|
||||
// ToProviderDefs converts tool definitions to provider-compatible format.
|
||||
|
||||
// This is the format expected by LLM provider APIs.
|
||||
|
||||
func (r *ToolRegistry) ToProviderDefs() []providers.ToolDefinition {
|
||||
r.mu.RLock()
|
||||
|
||||
defer r.mu.RUnlock()
|
||||
|
||||
sorted := r.sortedToolNames()
|
||||
|
||||
definitions := make([]providers.ToolDefinition, 0, len(sorted))
|
||||
|
||||
for _, name := range sorted {
|
||||
tool := r.tools[name]
|
||||
|
||||
schema := ToolToSchema(tool)
|
||||
|
||||
// Safely extract nested values with type checks
|
||||
|
||||
fn, ok := schema["function"].(map[string]any)
|
||||
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
|
||||
name, _ := fn["name"].(string)
|
||||
|
||||
desc, _ := fn["description"].(string)
|
||||
|
||||
params, _ := fn["parameters"].(map[string]any)
|
||||
|
||||
paramsRaw := json.RawMessage(`{}`)
|
||||
|
||||
if len(params) > 0 {
|
||||
if payload, err := json.Marshal(params); err == nil {
|
||||
paramsRaw = json.RawMessage(payload)
|
||||
}
|
||||
}
|
||||
|
||||
definitions = append(definitions, providers.ToolDefinition{
|
||||
Type: "function",
|
||||
|
||||
Function: providers.ToolFunctionDefinition{
|
||||
Name: name,
|
||||
|
||||
Description: desc,
|
||||
|
||||
Parameters: paramsRaw,
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
return definitions
|
||||
}
|
||||
|
||||
// List returns a list of all registered tool names.
|
||||
|
||||
func (r *ToolRegistry) List() []string {
|
||||
|
|
@ -291,101 +128,3 @@ func (r *ToolRegistry) Count() int {
|
|||
|
||||
return len(r.tools)
|
||||
}
|
||||
|
||||
// GetRuntimeStatus aggregates runtime status from all tools that implement StatusProvider.
|
||||
|
||||
// Returns empty string if no tool has status to report.
|
||||
|
||||
func (r *ToolRegistry) GetRuntimeStatus() string {
|
||||
r.mu.RLock()
|
||||
|
||||
defer r.mu.RUnlock()
|
||||
|
||||
var parts []string
|
||||
|
||||
for _, tool := range r.tools {
|
||||
if sp, ok := tool.(StatusProvider); ok {
|
||||
if s := sp.RuntimeStatus(); s != "" {
|
||||
parts = append(parts, s)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if len(parts) == 0 {
|
||||
return ""
|
||||
}
|
||||
|
||||
return strings.Join(parts, "\n\n")
|
||||
}
|
||||
|
||||
// buildParamHint extracts parameter names from a JSON schema and returns
|
||||
|
||||
// a hint string like "(task, label?, preset?)". Required params are bare,
|
||||
|
||||
// optional params have a trailing "?".
|
||||
|
||||
func buildParamHint(schema map[string]any) string {
|
||||
props, _ := schema["properties"].(map[string]any)
|
||||
|
||||
if len(props) == 0 {
|
||||
return ""
|
||||
}
|
||||
|
||||
reqSlice, _ := schema["required"].([]string)
|
||||
|
||||
reqSet := make(map[string]bool, len(reqSlice))
|
||||
|
||||
for _, r := range reqSlice {
|
||||
reqSet[r] = true
|
||||
}
|
||||
|
||||
names := make([]string, 0, len(props))
|
||||
|
||||
for name := range props {
|
||||
names = append(names, name)
|
||||
}
|
||||
|
||||
sort.Strings(names)
|
||||
|
||||
parts := make([]string, 0, len(names))
|
||||
|
||||
// Required params first, then optional
|
||||
|
||||
for _, name := range names {
|
||||
if reqSet[name] {
|
||||
parts = append(parts, name)
|
||||
}
|
||||
}
|
||||
|
||||
for _, name := range names {
|
||||
if !reqSet[name] {
|
||||
parts = append(parts, name+"?")
|
||||
}
|
||||
}
|
||||
|
||||
return "(" + strings.Join(parts, ", ") + ")"
|
||||
}
|
||||
|
||||
// GetSummaries returns human-readable summaries of all registered tools.
|
||||
|
||||
// Returns a slice of "- `name`(params) - description" strings.
|
||||
|
||||
func (r *ToolRegistry) GetSummaries() []string {
|
||||
r.mu.RLock()
|
||||
|
||||
defer r.mu.RUnlock()
|
||||
|
||||
sorted := r.sortedToolNames()
|
||||
|
||||
summaries := make([]string, 0, len(sorted))
|
||||
|
||||
for _, name := range sorted {
|
||||
tool := r.tools[name]
|
||||
|
||||
hint := buildParamHint(tool.Parameters())
|
||||
|
||||
summaries = append(summaries, fmt.Sprintf("- `%s`%s - %s", tool.Name(), hint, tool.Description()))
|
||||
}
|
||||
|
||||
return summaries
|
||||
}
|
||||
|
|
|
|||
267
pkg/tools/registry_ext.go
Normal file
267
pkg/tools/registry_ext.go
Normal file
|
|
@ -0,0 +1,267 @@
|
|||
package tools
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/sipeed/picoclaw/pkg/logger"
|
||||
"github.com/sipeed/picoclaw/pkg/providers"
|
||||
)
|
||||
|
||||
// ExecuteWithContext executes a tool with channel/chatID context and optional async callback.
|
||||
|
||||
// If the tool implements AsyncTool and a non-nil callback is provided,
|
||||
|
||||
// the callback will be set on the tool before execution.
|
||||
|
||||
func (r *ToolRegistry) ExecuteWithContext(
|
||||
ctx context.Context,
|
||||
|
||||
name string,
|
||||
|
||||
args map[string]any,
|
||||
|
||||
channel, chatID string,
|
||||
|
||||
asyncCallback AsyncCallback,
|
||||
) *ToolResult {
|
||||
logger.InfoCF("tool", "Tool execution started",
|
||||
|
||||
map[string]any{
|
||||
"tool": name,
|
||||
|
||||
"args": args,
|
||||
})
|
||||
|
||||
tool, ok := r.Get(name)
|
||||
|
||||
if !ok {
|
||||
available := strings.Join(r.List(), ", ")
|
||||
|
||||
logger.ErrorCF("tool", "Tool not found",
|
||||
|
||||
map[string]any{
|
||||
"tool": name,
|
||||
})
|
||||
|
||||
return ErrorResult(fmt.Sprintf(
|
||||
|
||||
"tool %q not found. Available tools: %s", name, available,
|
||||
)).WithError(fmt.Errorf("tool not found"))
|
||||
}
|
||||
|
||||
// If tool implements ContextualTool, set context
|
||||
|
||||
if contextualTool, ok := tool.(ContextualTool); ok && channel != "" && chatID != "" {
|
||||
contextualTool.SetContext(channel, chatID)
|
||||
}
|
||||
|
||||
// If tool implements AsyncTool and callback is provided, set callback
|
||||
|
||||
if asyncTool, ok := tool.(AsyncTool); ok && asyncCallback != nil {
|
||||
asyncTool.SetCallback(asyncCallback)
|
||||
|
||||
logger.DebugCF("tool", "Async callback injected",
|
||||
|
||||
map[string]any{
|
||||
"tool": name,
|
||||
})
|
||||
}
|
||||
|
||||
start := time.Now()
|
||||
|
||||
result := tool.Execute(ctx, args)
|
||||
|
||||
duration := time.Since(start)
|
||||
|
||||
// Log based on result type
|
||||
|
||||
if result.IsError {
|
||||
logger.ErrorCF("tool", "Tool execution failed",
|
||||
|
||||
map[string]any{
|
||||
"tool": name,
|
||||
|
||||
"duration": duration.Milliseconds(),
|
||||
|
||||
"error": result.ForLLM,
|
||||
})
|
||||
} else if result.Async {
|
||||
logger.InfoCF("tool", "Tool started (async)",
|
||||
|
||||
map[string]any{
|
||||
"tool": name,
|
||||
|
||||
"duration": duration.Milliseconds(),
|
||||
})
|
||||
} else {
|
||||
logger.InfoCF("tool", "Tool execution completed",
|
||||
|
||||
map[string]any{
|
||||
"tool": name,
|
||||
|
||||
"duration_ms": duration.Milliseconds(),
|
||||
|
||||
"result_length": len(result.ForLLM),
|
||||
})
|
||||
}
|
||||
|
||||
return result
|
||||
}
|
||||
|
||||
// ToProviderDefs converts tool definitions to provider-compatible format.
|
||||
|
||||
// This is the format expected by LLM provider APIs.
|
||||
|
||||
func (r *ToolRegistry) ToProviderDefs() []providers.ToolDefinition {
|
||||
r.mu.RLock()
|
||||
|
||||
defer r.mu.RUnlock()
|
||||
|
||||
sorted := r.sortedToolNames()
|
||||
|
||||
definitions := make([]providers.ToolDefinition, 0, len(sorted))
|
||||
|
||||
for _, name := range sorted {
|
||||
tool := r.tools[name]
|
||||
|
||||
schema := ToolToSchema(tool)
|
||||
|
||||
// Safely extract nested values with type checks
|
||||
|
||||
fn, ok := schema["function"].(map[string]any)
|
||||
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
|
||||
name, _ := fn["name"].(string)
|
||||
|
||||
desc, _ := fn["description"].(string)
|
||||
|
||||
params, _ := fn["parameters"].(map[string]any)
|
||||
|
||||
paramsRaw := json.RawMessage(`{}`)
|
||||
|
||||
if len(params) > 0 {
|
||||
if payload, err := json.Marshal(params); err == nil {
|
||||
paramsRaw = json.RawMessage(payload)
|
||||
}
|
||||
}
|
||||
|
||||
definitions = append(definitions, providers.ToolDefinition{
|
||||
Type: "function",
|
||||
|
||||
Function: providers.ToolFunctionDefinition{
|
||||
Name: name,
|
||||
|
||||
Description: desc,
|
||||
|
||||
Parameters: paramsRaw,
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
return definitions
|
||||
}
|
||||
|
||||
// GetRuntimeStatus aggregates runtime status from all tools that implement StatusProvider.
|
||||
|
||||
// Returns empty string if no tool has status to report.
|
||||
|
||||
func (r *ToolRegistry) GetRuntimeStatus() string {
|
||||
r.mu.RLock()
|
||||
|
||||
defer r.mu.RUnlock()
|
||||
|
||||
var parts []string
|
||||
|
||||
for _, tool := range r.tools {
|
||||
if sp, ok := tool.(StatusProvider); ok {
|
||||
if s := sp.RuntimeStatus(); s != "" {
|
||||
parts = append(parts, s)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if len(parts) == 0 {
|
||||
return ""
|
||||
}
|
||||
|
||||
return strings.Join(parts, "\n\n")
|
||||
}
|
||||
|
||||
// buildParamHint extracts parameter names from a JSON schema and returns
|
||||
|
||||
// a hint string like "(task, label?, preset?)". Required params are bare,
|
||||
|
||||
// optional params have a trailing "?".
|
||||
|
||||
func buildParamHint(schema map[string]any) string {
|
||||
props, _ := schema["properties"].(map[string]any)
|
||||
|
||||
if len(props) == 0 {
|
||||
return ""
|
||||
}
|
||||
|
||||
reqSlice, _ := schema["required"].([]string)
|
||||
|
||||
reqSet := make(map[string]bool, len(reqSlice))
|
||||
|
||||
for _, r := range reqSlice {
|
||||
reqSet[r] = true
|
||||
}
|
||||
|
||||
names := make([]string, 0, len(props))
|
||||
|
||||
for name := range props {
|
||||
names = append(names, name)
|
||||
}
|
||||
|
||||
sort.Strings(names)
|
||||
|
||||
parts := make([]string, 0, len(names))
|
||||
|
||||
// Required params first, then optional
|
||||
|
||||
for _, name := range names {
|
||||
if reqSet[name] {
|
||||
parts = append(parts, name)
|
||||
}
|
||||
}
|
||||
|
||||
for _, name := range names {
|
||||
if !reqSet[name] {
|
||||
parts = append(parts, name+"?")
|
||||
}
|
||||
}
|
||||
|
||||
return "(" + strings.Join(parts, ", ") + ")"
|
||||
}
|
||||
|
||||
// GetSummaries returns human-readable summaries of all registered tools.
|
||||
|
||||
// Returns a slice of "- `name`(params) - description" strings.
|
||||
|
||||
func (r *ToolRegistry) GetSummaries() []string {
|
||||
r.mu.RLock()
|
||||
|
||||
defer r.mu.RUnlock()
|
||||
|
||||
sorted := r.sortedToolNames()
|
||||
|
||||
summaries := make([]string, 0, len(sorted))
|
||||
|
||||
for _, name := range sorted {
|
||||
tool := r.tools[name]
|
||||
|
||||
hint := buildParamHint(tool.Parameters())
|
||||
|
||||
summaries = append(summaries, fmt.Sprintf("- `%s`%s - %s", tool.Name(), hint, tool.Description()))
|
||||
}
|
||||
|
||||
return summaries
|
||||
}
|
||||
Loading…
Add table
Reference in a new issue