From 89847f56721921d1ab5ffbe045b19d4610560c40 Mon Sep 17 00:00:00 2001 From: DBT Date: Mon, 23 Feb 2026 12:59:24 +0000 Subject: [PATCH] add subagents control tool and session kind visibility --- cmd/clawgo/cmd_status.go | 41 ++++++++++++++++++++ pkg/agent/loop.go | 1 + pkg/tools/subagent.go | 68 +++++++++++++++++++++++++-------- pkg/tools/subagents_tool.go | 76 +++++++++++++++++++++++++++++++++++++ 4 files changed, 170 insertions(+), 16 deletions(-) create mode 100644 pkg/tools/subagents_tool.go diff --git a/cmd/clawgo/cmd_status.go b/cmd/clawgo/cmd_status.go index 8284e9f..250bc38 100644 --- a/cmd/clawgo/cmd_status.go +++ b/cmd/clawgo/cmd_status.go @@ -1,6 +1,7 @@ package main import ( + "encoding/json" "fmt" "os" "path/filepath" @@ -90,5 +91,45 @@ func statusCmd() { if data, err := os.ReadFile(triggerStats); err == nil { fmt.Printf("Trigger Stats: %s\n", strings.TrimSpace(string(data))) } + + sessionsDir := filepath.Join(filepath.Dir(configPath), "sessions") + if kinds, err := collectSessionKindCounts(sessionsDir); err == nil && len(kinds) > 0 { + fmt.Println("Session Kinds:") + for _, k := range []string{"main", "cron", "subagent", "hook", "node", "other"} { + if v, ok := kinds[k]; ok { + fmt.Printf(" %s: %d\n", k, v) + } + } + } } } + +func collectSessionKindCounts(sessionsDir string) (map[string]int, error) { + entries, err := os.ReadDir(sessionsDir) + if err != nil { + return nil, err + } + counts := map[string]int{} + for _, e := range entries { + if e.IsDir() || !strings.HasSuffix(e.Name(), ".meta") { + continue + } + metaPath := filepath.Join(sessionsDir, e.Name()) + data, err := os.ReadFile(metaPath) + if err != nil { + continue + } + var meta struct { + Kind string `json:"kind"` + } + if err := json.Unmarshal(data, &meta); err != nil { + continue + } + kind := strings.TrimSpace(strings.ToLower(meta.Kind)) + if kind == "" { + kind = "other" + } + counts[kind]++ + } + return counts, nil +} diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index 06e4b8a..a142c16 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -97,6 +97,7 @@ func NewAgentLoop(cfg *config.Config, msgBus *bus.MessageBus, provider providers subagentManager := tools.NewSubagentManager(provider, workspace, msgBus, orchestrator) spawnTool := tools.NewSpawnTool(subagentManager) toolsRegistry.Register(spawnTool) + toolsRegistry.Register(tools.NewSubagentsTool(subagentManager)) // Register edit file tool editFileTool := tools.NewEditFileTool(workspace) diff --git a/pkg/tools/subagent.go b/pkg/tools/subagent.go index 4410658..6ecb975 100644 --- a/pkg/tools/subagent.go +++ b/pkg/tools/subagent.go @@ -23,27 +23,30 @@ type SubagentTask struct { Status string Result string Created int64 + Updated int64 } type SubagentManager struct { - tasks map[string]*SubagentTask - mu sync.RWMutex - provider providers.LLMProvider - bus *bus.MessageBus - orc *Orchestrator - workspace string - nextID int - runFunc SubagentRunFunc + tasks map[string]*SubagentTask + cancelFuncs map[string]context.CancelFunc + mu sync.RWMutex + provider providers.LLMProvider + bus *bus.MessageBus + orc *Orchestrator + workspace string + nextID int + runFunc SubagentRunFunc } func NewSubagentManager(provider providers.LLMProvider, workspace string, bus *bus.MessageBus, orc *Orchestrator) *SubagentManager { return &SubagentManager{ - tasks: make(map[string]*SubagentTask), - provider: provider, - bus: bus, - orc: orc, - workspace: workspace, - nextID: 1, + tasks: make(map[string]*SubagentTask), + cancelFuncs: make(map[string]context.CancelFunc), + provider: provider, + bus: bus, + orc: orc, + workspace: workspace, + nextID: 1, } } @@ -54,6 +57,7 @@ func (sm *SubagentManager) Spawn(ctx context.Context, task, label, originChannel taskID := fmt.Sprintf("subagent-%d", sm.nextID) sm.nextID++ + now := time.Now().UnixMilli() subagentTask := &SubagentTask{ ID: taskID, Task: task, @@ -63,11 +67,14 @@ func (sm *SubagentManager) Spawn(ctx context.Context, task, label, originChannel OriginChannel: originChannel, OriginChatID: originChatID, Status: "running", - Created: time.Now().UnixMilli(), + Created: now, + Updated: now, } + taskCtx, cancel := context.WithCancel(ctx) sm.tasks[taskID] = subagentTask + sm.cancelFuncs[taskID] = cancel - go sm.runTask(ctx, subagentTask) + go sm.runTask(taskCtx, subagentTask) desc := fmt.Sprintf("Spawned subagent for task: %s", task) if label != "" { @@ -80,9 +87,16 @@ func (sm *SubagentManager) Spawn(ctx context.Context, task, label, originChannel } func (sm *SubagentManager) runTask(ctx context.Context, task *SubagentTask) { + defer func() { + sm.mu.Lock() + delete(sm.cancelFuncs, task.ID) + sm.mu.Unlock() + }() + sm.mu.Lock() task.Status = "running" task.Created = time.Now().UnixMilli() + task.Updated = task.Created sm.mu.Unlock() if sm.orc != nil && task.PipelineID != "" && task.PipelineTask != "" { @@ -100,12 +114,14 @@ func (sm *SubagentManager) runTask(ctx context.Context, task *SubagentTask) { if err != nil { task.Status = "failed" task.Result = fmt.Sprintf("Error: %v", err) + task.Updated = time.Now().UnixMilli() if sm.orc != nil && task.PipelineID != "" && task.PipelineTask != "" { _ = sm.orc.MarkTaskDone(task.PipelineID, task.PipelineTask, task.Result, err) } } else { task.Status = "completed" task.Result = result + task.Updated = time.Now().UnixMilli() if sm.orc != nil && task.PipelineID != "" && task.PipelineTask != "" { _ = sm.orc.MarkTaskDone(task.PipelineID, task.PipelineTask, task.Result, nil) } @@ -132,12 +148,14 @@ func (sm *SubagentManager) runTask(ctx context.Context, task *SubagentTask) { if err != nil { task.Status = "failed" task.Result = fmt.Sprintf("Error: %v", err) + task.Updated = time.Now().UnixMilli() if sm.orc != nil && task.PipelineID != "" && task.PipelineTask != "" { _ = sm.orc.MarkTaskDone(task.PipelineID, task.PipelineTask, task.Result, err) } } else { task.Status = "completed" task.Result = response.Content + task.Updated = time.Now().UnixMilli() if sm.orc != nil && task.PipelineID != "" && task.PipelineTask != "" { _ = sm.orc.MarkTaskDone(task.PipelineID, task.PipelineTask, task.Result, nil) } @@ -194,3 +212,21 @@ func (sm *SubagentManager) ListTasks() []*SubagentTask { } return tasks } + +func (sm *SubagentManager) KillTask(taskID string) bool { + sm.mu.Lock() + defer sm.mu.Unlock() + t, ok := sm.tasks[taskID] + if !ok { + return false + } + if cancel, ok := sm.cancelFuncs[taskID]; ok { + cancel() + delete(sm.cancelFuncs, taskID) + } + if t.Status == "running" { + t.Status = "killed" + t.Updated = time.Now().UnixMilli() + } + return true +} diff --git a/pkg/tools/subagents_tool.go b/pkg/tools/subagents_tool.go new file mode 100644 index 0000000..1785bfb --- /dev/null +++ b/pkg/tools/subagents_tool.go @@ -0,0 +1,76 @@ +package tools + +import ( + "context" + "fmt" + "strings" +) + +type SubagentsTool struct { + manager *SubagentManager +} + +func NewSubagentsTool(m *SubagentManager) *SubagentsTool { + return &SubagentsTool{manager: m} +} + +func (t *SubagentsTool) Name() string { return "subagents" } + +func (t *SubagentsTool) Description() string { + return "Manage subagent runs in current process: list, info, kill" +} + +func (t *SubagentsTool) Parameters() map[string]interface{} { + return map[string]interface{}{ + "type": "object", + "properties": map[string]interface{}{ + "action": map[string]interface{}{"type": "string", "description": "list|info|kill"}, + "id": map[string]interface{}{"type": "string", "description": "subagent id for info/kill"}, + }, + "required": []string{"action"}, + } +} + +func (t *SubagentsTool) Execute(ctx context.Context, args map[string]interface{}) (string, error) { + _ = ctx + if t.manager == nil { + return "subagent manager not available", nil + } + action, _ := args["action"].(string) + action = strings.ToLower(strings.TrimSpace(action)) + id, _ := args["id"].(string) + id = strings.TrimSpace(id) + + switch action { + case "list": + tasks := t.manager.ListTasks() + if len(tasks) == 0 { + return "No subagents.", nil + } + var sb strings.Builder + sb.WriteString("Subagents:\n") + for _, task := range tasks { + sb.WriteString(fmt.Sprintf("- %s [%s] label=%s\n", task.ID, task.Status, task.Label)) + } + return strings.TrimSpace(sb.String()), nil + case "info": + if id == "" { + return "id is required for info", nil + } + task, ok := t.manager.GetTask(id) + if !ok { + return "subagent not found", nil + } + return fmt.Sprintf("ID: %s\nStatus: %s\nLabel: %s\nTask: %s\nResult:\n%s", task.ID, task.Status, task.Label, task.Task, task.Result), nil + case "kill": + if id == "" { + return "id is required for kill", nil + } + if !t.manager.KillTask(id) { + return "subagent not found", nil + } + return "subagent kill requested", nil + default: + return "unsupported action", nil + } +}