From 2c1ef253926f6d0e76632c1141d2ed4edafc5094 Mon Sep 17 00:00:00 2001 From: viettranx Date: Tue, 31 Mar 2026 11:45:16 +0700 Subject: [PATCH] feat(subagent): token tracking, edition limits, waitAll, auto-retry, producer-consumer announce (#600) - Token cost tracking: accumulate input/output tokens per subagent, include in announce messages and persist to DB - Per-edition rate limits: MaxSubagentConcurrent/Depth on Edition struct, tenant-scoped concurrency enforcement in Spawn/RunSync - WaitAll action: spawn(action=wait, timeout=N) blocks until all children complete, returns merged summary - Auto-retry: configurable MaxRetries (default 2) with linear backoff for transient LLM failures - Producer-consumer announce queue: merges staggered subagent results into single LLM run (same pattern as team task announces) - Raw metadata in bus messages to prevent double-formatting - Fire-and-forget DB persistence with detached context + tenant scope - Split oversized files for <200 line compliance --- cmd/gateway.go | 7 +- cmd/gateway_consumer.go | 3 +- cmd/gateway_consumer_deps.go | 1 + cmd/gateway_consumer_handlers.go | 120 ++++------- cmd/gateway_subagent_announce_format.go | 70 ++++++ cmd/gateway_subagent_announce_queue.go | 202 ++++++++++++++++++ internal/config/config.go | 1 + internal/edition/edition.go | 40 ++-- internal/tools/announce_queue.go | 28 +-- internal/tools/subagent_config.go | 1 + internal/tools/subagent_control.go | 31 +++ internal/tools/subagent_exec.go | 75 +++++-- internal/tools/subagent_persist.go | 97 +++++++++ internal/tools/subagent_spawn.go | 190 ++++++++++++++++ internal/tools/subagent_spawn_tool.go | 111 +--------- internal/tools/subagent_spawn_tool_actions.go | 156 ++++++++++++++ internal/tools/team_metadata_keys.go | 8 +- 17 files changed, 913 insertions(+), 228 deletions(-) create mode 100644 cmd/gateway_subagent_announce_format.go create mode 100644 cmd/gateway_subagent_announce_queue.go create mode 100644 internal/tools/subagent_persist.go create mode 100644 internal/tools/subagent_spawn.go create mode 100644 internal/tools/subagent_spawn_tool_actions.go diff --git a/cmd/gateway.go b/cmd/gateway.go index 985a179b..ff39bb2e 100644 --- a/cmd/gateway.go +++ b/cmd/gateway.go @@ -219,6 +219,9 @@ func runGateway() { }, ) subagentMgr.SetAnnounceQueue(announceQueue) + if pgStores.SubagentTasks != nil { + subagentMgr.SetTaskStore(pgStores.SubagentTasks) + } toolsReg.Register(tools.NewSpawnTool(subagentMgr, "default", 0)) slog.Info("subagent system enabled", "tools", []string{"spawn"}) @@ -552,7 +555,7 @@ func runGateway() { instanceLoader = channels.NewInstanceLoader(pgStores.ChannelInstances, pgStores.Agents, channelMgr, msgBus, pgStores.Pairing) instanceLoader.SetProviderRegistry(providerRegistry) instanceLoader.SetPendingCompactionConfig(cfg.Channels.PendingCompaction) - instanceLoader.RegisterFactory(channels.TypeTelegram, telegram.FactoryWithStores(pgStores.Agents, pgStores.ConfigPermissions, pgStores.Teams, pgStores.PendingMessages)) + instanceLoader.RegisterFactory(channels.TypeTelegram, telegram.FactoryWithStores(pgStores.Agents, pgStores.ConfigPermissions, pgStores.Teams, pgStores.SubagentTasks, pgStores.PendingMessages)) instanceLoader.RegisterFactory(channels.TypeDiscord, discord.FactoryWithStores(pgStores.Agents, pgStores.ConfigPermissions, pgStores.PendingMessages)) instanceLoader.RegisterFactory(channels.TypeFeishu, feishu.FactoryWithPendingStore(pgStores.PendingMessages)) instanceLoader.RegisterFactory(channels.TypeZaloOA, zalo.Factory) @@ -1087,7 +1090,7 @@ func runGateway() { channelMgr.SetContactCollector(contactCollector) // propagate to all channel handlers } - go consumeInboundMessages(ctx, msgBus, agentRouter, cfg, sched, channelMgr, consumerTeamStore, quotaChecker, pgStores.Sessions, pgStores.Agents, contactCollector, postTurn) + go consumeInboundMessages(ctx, msgBus, agentRouter, cfg, sched, channelMgr, consumerTeamStore, quotaChecker, pgStores.Sessions, pgStores.Agents, contactCollector, postTurn, subagentMgr) // Task recovery ticker: re-dispatches stale/pending team tasks on startup and periodically. var taskTicker *tasks.TaskTicker diff --git a/cmd/gateway_consumer.go b/cmd/gateway_consumer.go index 0690bf6b..3c88bdc9 100644 --- a/cmd/gateway_consumer.go +++ b/cmd/gateway_consumer.go @@ -25,7 +25,7 @@ import ( // and routes them through the scheduler/agent loop, then publishes the response back. // Also handles subagent announcements: routes them through the parent agent's session // (matching TS subagent-announce.ts pattern) so the agent can reformulate for the user. -func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents *agent.Router, cfg *config.Config, sched *scheduler.Scheduler, channelMgr *channels.Manager, teamStore store.TeamStore, quotaChecker *channels.QuotaChecker, sessStore store.SessionStore, agentStore store.AgentStore, contactCollector *store.ContactCollector, postTurn tools.PostTurnProcessor) { +func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents *agent.Router, cfg *config.Config, sched *scheduler.Scheduler, channelMgr *channels.Manager, teamStore store.TeamStore, quotaChecker *channels.QuotaChecker, sessStore store.SessionStore, agentStore store.AgentStore, contactCollector *store.ContactCollector, postTurn tools.PostTurnProcessor, subagentMgr *tools.SubagentManager) { slog.Info("inbound message consumer started") // Inbound message deduplication (matching TS src/infra/dedupe.ts + inbound-dedupe.ts). @@ -56,6 +56,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents PostTurn: postTurn, QuotaChecker: quotaChecker, ContactCollector: contactCollector, + SubagentMgr: subagentMgr, GetAnnounceMu: getAnnounceMu, } diff --git a/cmd/gateway_consumer_deps.go b/cmd/gateway_consumer_deps.go index 76a13c6f..faf3755c 100644 --- a/cmd/gateway_consumer_deps.go +++ b/cmd/gateway_consumer_deps.go @@ -27,6 +27,7 @@ type ConsumerDeps struct { QuotaChecker *channels.QuotaChecker ContactCollector *store.ContactCollector TaskRunSessions sync.Map + SubagentMgr *tools.SubagentManager BgWg sync.WaitGroup GetAnnounceMu func(string) *sync.Mutex } diff --git a/cmd/gateway_consumer_handlers.go b/cmd/gateway_consumer_handlers.go index a0124b94..5caf19e9 100644 --- a/cmd/gateway_consumer_handlers.go +++ b/cmd/gateway_consumer_handlers.go @@ -2,11 +2,12 @@ package cmd import ( "context" - "errors" "fmt" "log/slog" "path/filepath" + "strconv" "strings" + "time" "github.com/google/uuid" @@ -85,91 +86,60 @@ func handleSubagentAnnounce( announceUserID = fmt.Sprintf("group:%s:%s", origChannel, msg.ChatID) } - // Build outbound metadata for topic/thread routing. - outMeta := buildAnnounceOutMeta(origLocalKey) + // Build announce entry from raw metadata (avoids double-formatting). + runtimeMs, _ := strconv.ParseInt(msg.Metadata[tools.MetaSubagentRuntime], 10, 64) + iterations, _ := strconv.Atoi(msg.Metadata[tools.MetaSubagentIterations]) + inputToks, _ := strconv.ParseInt(msg.Metadata[tools.MetaSubagentInputToks], 10, 64) + outputToks, _ := strconv.ParseInt(msg.Metadata[tools.MetaSubagentOutputToks], 10, 64) - // Build request before goroutine to capture msg fields. - // WS channel has no outbound handler — media converted to markdown URLs - // and appended to the assistant response via ContentSuffix, which the - // agent loop applies BEFORE saving to session and emitting run.completed. - fwdMedia := msg.Media - contentSuffix := "" - if origChannel == "ws" && len(msg.Media) > 0 { - contentSuffix = mediaToMarkdownFromPaths(msg.Media, deps.Cfg) - fwdMedia = nil // WS: images delivered via ContentSuffix, not ForwardMedia + // Use raw result from metadata if available; fall back to formatted Content for backward compat. + rawResult := msg.Metadata[tools.MetaSubagentResult] + if rawResult == "" { + rawResult = msg.Content } - announceReq := agent.RunRequest{ + entry := subagentAnnounceEntry{ + Label: msg.Metadata[tools.MetaSubagentLabel], + Status: msg.Metadata[tools.MetaSubagentStatus], + Content: rawResult, + Media: msg.Media, + InputTokens: inputToks, + OutputTokens: outputToks, + Runtime: time.Duration(runtimeMs) * time.Millisecond, + Iterations: iterations, + } + + queueKey := fmt.Sprintf("%s:%s", msg.TenantID, sessionKey) + routing := subagentAnnounceRouting{ + QueueKey: queueKey, SessionKey: sessionKey, - Message: msg.Content, - ForwardMedia: fwdMedia, - ContentSuffix: contentSuffix, - Channel: origChannel, - ChannelType: origChannelType, - ChatID: msg.ChatID, - PeerKind: origPeerKind, - LocalKey: origLocalKey, + TenantID: msg.TenantID, + OrigChannel: origChannel, + OrigChannelType: origChannelType, + OrigChatID: msg.ChatID, + OrigPeerKind: origPeerKind, + OrigLocalKey: origLocalKey, UserID: announceUserID, - RunID: fmt.Sprintf("announce-%s", msg.SenderID), - RunKind: "announce", - HideInput: true, // don't persist raw system message in chat history - Stream: false, + ParentAgent: parentAgent, ParentTraceID: parentTraceID, ParentRootSpanID: parentRootSpanID, + OutMeta: buildAnnounceOutMeta(origLocalKey), } - // Handle announce asynchronously with per-session serialization. - // The mutex ensures concurrent announces for the same session wait for - // each other, so each reads up-to-date session history. - deps.BgWg.Add(1) - go func(sessionKey, origCh, chatID, senderID, label string, meta map[string]string, req agent.RunRequest) { - defer deps.BgWg.Done() - defer safego.Recover(nil, "component", "subagent_announce", "session", sessionKey) - mu := deps.GetAnnounceMu(sessionKey) - mu.Lock() - defer mu.Unlock() - outCh := deps.Sched.Schedule(ctx, scheduler.LaneSubagent, req) - outcome := <-outCh - if outcome.Err != nil { - if errors.Is(outcome.Err, context.Canceled) { - slog.Info("subagent announce: run cancelled", "subagent", senderID) - return - } - slog.Error("subagent announce: agent run failed", "error", outcome.Err) - deps.MsgBus.PublishOutbound(bus.OutboundMessage{ - Channel: origCh, - ChatID: chatID, - Content: formatAgentError(outcome.Err), - Metadata: meta, - }) - return - } + // Enqueue into producer-consumer queue using tenant-scoped key from routing. + q, isProcessor := enqueueSubagentAnnounce(queueKey, entry) + if isProcessor { + deps.BgWg.Add(1) + go func() { + defer deps.BgWg.Done() + defer safego.Recover(nil, "component", "subagent_announce_loop", "session", sessionKey) - // Suppress empty/NO_REPLY (matching TS normalize-reply.ts / tokens.ts). - isSilent := outcome.Result.Content == "" || agent.IsSilentReply(outcome.Result.Content) - if isSilent && len(outcome.Result.Media) == 0 { - slog.Info("subagent announce: suppressed silent/empty reply", - "subagent", senderID, - "label", label, - ) - return - } + // Fetch live roster for merged announce context. + roster := deps.SubagentMgr.RosterForParent(parentAgent) - // Deliver agent's reformulated response to origin channel. - announceContent := outcome.Result.Content - if isSilent { - announceContent = "" // suppress NO_REPLY text but still send media - } - - outMsg := bus.OutboundMessage{ - Channel: origCh, - ChatID: chatID, - Content: announceContent, - Metadata: meta, - } - appendMediaToOutbound(&outMsg, outcome.Result.Media) - deps.MsgBus.PublishOutbound(outMsg) - }(sessionKey, origChannel, msg.ChatID, msg.SenderID, msg.Metadata[tools.MetaSubagentLabel], outMeta, announceReq) + processSubagentAnnounceLoop(ctx, q, routing, roster, deps.SubagentMgr, deps.Sched, deps.MsgBus, deps.Cfg) + }() + } return true } diff --git a/cmd/gateway_subagent_announce_format.go b/cmd/gateway_subagent_announce_format.go new file mode 100644 index 00000000..32cd6051 --- /dev/null +++ b/cmd/gateway_subagent_announce_format.go @@ -0,0 +1,70 @@ +package cmd + +import ( + "fmt" + "strings" + "time" + + "github.com/nextlevelbuilder/goclaw/internal/tools" +) + +// buildMergedSubagentAnnounce creates the announce message for one or more subagent results. +func buildMergedSubagentAnnounce(entries []subagentAnnounceEntry, roster tools.SubagentRoster) string { + var sb strings.Builder + + if len(entries) == 1 { + e := entries[0] + statusLabel := "completed successfully" + if e.Status == "failed" { + statusLabel = "failed" + } else if e.Status == "cancelled" { + statusLabel = "was cancelled" + } + fmt.Fprintf(&sb, "[System Message] A subagent task %q just %s.\n\nResult:\n%s\n\nStats: runtime %s, iterations %d, tokens %d in / %d out\n", + e.Label, statusLabel, e.Content, + e.Runtime.Round(time.Millisecond), e.Iterations, + e.InputTokens, e.OutputTokens) + } else { + completed, failed := 0, 0 + for _, e := range entries { + switch e.Status { + case "completed": + completed++ + case "failed": + failed++ + } + } + if failed > 0 && completed > 0 { + fmt.Fprintf(&sb, "[System Message] %d subagent task(s) completed, %d failed.\n", completed, failed) + } else if failed > 0 { + fmt.Fprintf(&sb, "[System Message] %d subagent task(s) failed.\n", failed) + } else { + fmt.Fprintf(&sb, "[System Message] %d subagent tasks completed.\n", completed) + } + for i, e := range entries { + fmt.Fprintf(&sb, "\n--- Task #%d: %q (%s, %s, tokens %d/%d) ---\nResult: %s\n", + i+1, e.Label, e.Status, + e.Runtime.Round(time.Millisecond), + e.InputTokens, e.OutputTokens, e.Content) + } + } + + // Append roster for context. + sb.WriteString("\n") + sb.WriteString(tools.BuildReplyInstruction(roster)) + + // Smart prompting based on remaining tasks. + running := 0 + for _, e := range roster.Entries { + if e.Status == tools.TaskStatusRunning { + running++ + } + } + if running == 0 { + sb.WriteString("\n\nAll subagent tasks finished. Present a comprehensive summary of ALL results to the user.") + } else { + sb.WriteString("\n\nSome subagents are still running. Briefly acknowledge this result (1-2 sentences). A full summary will come when all tasks complete.") + } + + return sb.String() +} diff --git a/cmd/gateway_subagent_announce_queue.go b/cmd/gateway_subagent_announce_queue.go new file mode 100644 index 00000000..83a55746 --- /dev/null +++ b/cmd/gateway_subagent_announce_queue.go @@ -0,0 +1,202 @@ +package cmd + +import ( + "context" + "errors" + "fmt" + "log/slog" + "sync" + "time" + + "github.com/google/uuid" + + "github.com/nextlevelbuilder/goclaw/internal/agent" + "github.com/nextlevelbuilder/goclaw/internal/bus" + "github.com/nextlevelbuilder/goclaw/internal/config" + "github.com/nextlevelbuilder/goclaw/internal/scheduler" + "github.com/nextlevelbuilder/goclaw/internal/store" + "github.com/nextlevelbuilder/goclaw/internal/tools" +) + +// subagentAnnounceEntry holds one subagent completion result waiting to be announced. +type subagentAnnounceEntry struct { + Label string + Status string // "completed", "failed", "cancelled" + Content string + Media []bus.MediaFile + InputTokens int64 + OutputTokens int64 + Runtime time.Duration + Iterations int +} + +// subagentAnnounceRouting holds shared routing info captured by the first enqueue. +type subagentAnnounceRouting struct { + QueueKey string // tenant-scoped key for sync.Map (tenantID:sessionKey) + SessionKey string // original session key (no tenant prefix) for RunRequest + TenantID uuid.UUID // preserved for tenant-scoped scheduling + OrigChannel string + OrigChannelType string + OrigChatID string + OrigPeerKind string + OrigLocalKey string + UserID string + ParentAgent string + ParentTraceID uuid.UUID + ParentRootSpanID uuid.UUID + OutMeta map[string]string +} + +// subagentAnnounceQueue is a producer-consumer queue per parent session. +// Multiple subagent goroutines enqueue entries; one processor drains and merges. +type subagentAnnounceQueue struct { + mu sync.Mutex + running bool + entries []subagentAnnounceEntry +} + +// subagentAnnounceQueues maps sessionKey → queue. Cleaned up when queue finishes. +var subagentAnnounceQueues sync.Map + +func getOrCreateSubagentAnnounceQueue(key string) *subagentAnnounceQueue { + v, _ := subagentAnnounceQueues.LoadOrStore(key, &subagentAnnounceQueue{}) + return v.(*subagentAnnounceQueue) +} + +// enqueueSubagentAnnounce adds a result to the queue. Returns (queue, isProcessor). +// If isProcessor=true, the caller must run processSubagentAnnounceLoop. +func enqueueSubagentAnnounce(key string, entry subagentAnnounceEntry) (*subagentAnnounceQueue, bool) { + q := getOrCreateSubagentAnnounceQueue(key) + q.mu.Lock() + defer q.mu.Unlock() + q.entries = append(q.entries, entry) + if q.running { + return q, false + } + q.running = true + return q, true +} + +func (q *subagentAnnounceQueue) drain() []subagentAnnounceEntry { + q.mu.Lock() + defer q.mu.Unlock() + out := q.entries + q.entries = nil + return out +} + +// tryFinish atomically checks for pending entries and marks the queue idle. +func (q *subagentAnnounceQueue) tryFinish(key string) bool { + q.mu.Lock() + defer q.mu.Unlock() + if len(q.entries) > 0 { + return false + } + q.running = false + subagentAnnounceQueues.Delete(key) + return true +} + +// processSubagentAnnounceLoop drains entries, builds merged announce, schedules to parent. +func processSubagentAnnounceLoop( + ctx context.Context, + q *subagentAnnounceQueue, + r subagentAnnounceRouting, + roster tools.SubagentRoster, + subagentMgr *tools.SubagentManager, + sched *scheduler.Scheduler, + msgBus *bus.MessageBus, + cfg *config.Config, +) { + // Ensure tenant scope is always set for the scheduler. + if r.TenantID != uuid.Nil { + ctx = store.WithTenantID(ctx, r.TenantID) + } + + for { + select { + case <-ctx.Done(): + q.tryFinish(r.QueueKey) + return + default: + } + + entries := q.drain() + if len(entries) == 0 { + if q.tryFinish(r.QueueKey) { + return + } + // Brief sleep to avoid tight spin when entries arrive between drain and tryFinish. + time.Sleep(50 * time.Millisecond) + continue + } + + // Refresh roster each iteration for up-to-date task statuses. + roster = subagentMgr.RosterForParent(r.ParentAgent) + content := buildMergedSubagentAnnounce(entries, roster) + + // Collect media from all entries. + var fwdMedia []bus.MediaFile + for _, e := range entries { + fwdMedia = append(fwdMedia, e.Media...) + } + contentSuffix := "" + if r.OrigChannel == "ws" && len(fwdMedia) > 0 { + contentSuffix = mediaToMarkdownFromPaths(fwdMedia, cfg) + fwdMedia = nil + } + + req := agent.RunRequest{ + SessionKey: r.SessionKey, + Message: content, + ForwardMedia: fwdMedia, + ContentSuffix: contentSuffix, + Channel: r.OrigChannel, + ChannelType: r.OrigChannelType, + ChatID: r.OrigChatID, + PeerKind: r.OrigPeerKind, + LocalKey: r.OrigLocalKey, + UserID: r.UserID, + RunID: fmt.Sprintf("subagent-announce-%s-%d", r.ParentAgent, len(entries)), + RunKind: "announce", + HideInput: true, + Stream: false, + ParentTraceID: r.ParentTraceID, + ParentRootSpanID: r.ParentRootSpanID, + } + + outCh := sched.Schedule(ctx, scheduler.LaneSubagent, req) + outcome := <-outCh + + if outcome.Err != nil { + if !errors.Is(outcome.Err, context.Canceled) { + slog.Error("subagent announce: lead run failed", "error", outcome.Err, "batch_size", len(entries)) + msgBus.PublishOutbound(bus.OutboundMessage{ + Channel: r.OrigChannel, + ChatID: r.OrigChatID, + Content: formatAgentError(outcome.Err), + Metadata: r.OutMeta, + }) + } + } else { + isSilent := outcome.Result.Content == "" || agent.IsSilentReply(outcome.Result.Content) + if !(isSilent && len(outcome.Result.Media) == 0) { + out := outcome.Result.Content + if isSilent { + out = "" + } + outMsg := bus.OutboundMessage{ + Channel: r.OrigChannel, + ChatID: r.OrigChatID, + Content: out, + Metadata: r.OutMeta, + } + appendMediaToOutbound(&outMsg, outcome.Result.Media) + msgBus.PublishOutbound(outMsg) + } + } + + slog.Info("subagent announce: batch processed", + "batch_size", len(entries), "session", r.SessionKey) + } +} diff --git a/internal/config/config.go b/internal/config/config.go index e7411103..e995915d 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -350,6 +350,7 @@ type SubagentsConfig struct { MaxSpawnDepth int `json:"maxSpawnDepth,omitempty"` // default 1, range 1-5 MaxChildrenPerAgent int `json:"maxChildrenPerAgent,omitempty"` // default 5, range 1-20 ArchiveAfterMinutes int `json:"archiveAfterMinutes,omitempty"` // default 60 + MaxRetries int `json:"maxRetries,omitempty"` // max LLM retries on error (default 2) Model string `json:"model,omitempty"` // model override for subagents } diff --git a/internal/edition/edition.go b/internal/edition/edition.go index 73918b91..b93de9d1 100644 --- a/internal/edition/edition.go +++ b/internal/edition/edition.go @@ -7,15 +7,17 @@ import "sync/atomic" // Edition defines the feature limits for a GoClaw instance. type Edition struct { - Name string `json:"name"` // "standard" or "lite" - MaxAgents int `json:"max_agents"` // 0 = unlimited - MaxTeams int `json:"max_teams"` // 0 = unlimited - MaxTeamMembers int `json:"max_team_members"` // 0 = unlimited - MaxChannels map[string]int `json:"max_channels"` // per channel type, nil = unlimited - KGEnabled bool `json:"kg_enabled"` - RBACEnabled bool `json:"rbac_enabled"` - TeamFullMode bool `json:"team_full_mode"` // false = lite task actions only - VectorSearch bool `json:"vector_search"` // false = FTS5 only + Name string `json:"name"` // "standard" or "lite" + MaxAgents int `json:"max_agents"` // 0 = unlimited + MaxTeams int `json:"max_teams"` // 0 = unlimited + MaxTeamMembers int `json:"max_team_members"` // 0 = unlimited + MaxChannels map[string]int `json:"max_channels"` // per channel type, nil = unlimited + MaxSubagentConcurrent int `json:"max_subagent_concurrent"` // 0 = unlimited + MaxSubagentDepth int `json:"max_subagent_depth"` // 0 = use config default + KGEnabled bool `json:"kg_enabled"` + RBACEnabled bool `json:"rbac_enabled"` + TeamFullMode bool `json:"team_full_mode"` // false = lite task actions only + VectorSearch bool `json:"vector_search"` // false = FTS5 only } // --- Presets --- @@ -31,15 +33,17 @@ var Standard = Edition{ // Lite is the desktop/self-hosted edition with sensible limits. var Lite = Edition{ - Name: "lite", - MaxAgents: 5, - MaxTeams: 1, - MaxTeamMembers: 5, - MaxChannels: map[string]int{"telegram": 1, "discord": 1}, - KGEnabled: false, - RBACEnabled: false, - TeamFullMode: false, - VectorSearch: false, + Name: "lite", + MaxAgents: 5, + MaxTeams: 1, + MaxTeamMembers: 5, + MaxChannels: map[string]int{"telegram": 1, "discord": 1}, + MaxSubagentConcurrent: 2, + MaxSubagentDepth: 1, + KGEnabled: false, + RBACEnabled: false, + TeamFullMode: false, + VectorSearch: false, } // --- Global state --- diff --git a/internal/tools/announce_queue.go b/internal/tools/announce_queue.go index 6b7cce6f..becda285 100644 --- a/internal/tools/announce_queue.go +++ b/internal/tools/announce_queue.go @@ -13,13 +13,15 @@ import ( // AnnounceQueueItem represents a single subagent result waiting to be announced. type AnnounceQueueItem struct { - SubagentID string - Label string - Status string // "completed", "failed", "cancelled" - Result string - Media []bus.MediaFile // media files from tool results - Runtime time.Duration - Iterations int + SubagentID string + Label string + Status string // "completed", "failed", "cancelled" + Result string + Media []bus.MediaFile // media files from tool results + Runtime time.Duration + Iterations int + InputTokens int64 + OutputTokens int64 } // AnnounceMetadata carries origin info for routing the batched announce. @@ -140,15 +142,16 @@ func FormatBatchedAnnounce(items []AnnounceQueueItem, roster SubagentRoster) str statusLabel = "was cancelled" } - replyInstruction := buildReplyInstruction(roster) + replyInstruction := BuildReplyInstruction(roster) return fmt.Sprintf( "[System Message] A subagent task %q just %s.\n\n"+ "Result:\n%s\n\n"+ - "Stats: runtime %s, iterations %d\n\n"+ + "Stats: runtime %s, iterations %d, tokens %d in / %d out\n\n"+ "%s", item.Label, statusLabel, item.Result, item.Runtime.Round(time.Millisecond), item.Iterations, + item.InputTokens, item.OutputTokens, replyInstruction, ) } @@ -166,22 +169,23 @@ func FormatBatchedAnnounce(items []AnnounceQueueItem, roster SubagentRoster) str } sb.WriteString(fmt.Sprintf( - "\n---\nTask #%d: %q %s (runtime %s, iterations %d)\nResult: %s\n", + "\n---\nTask #%d: %q %s (runtime %s, iterations %d, tokens %d/%d)\nResult: %s\n", i+1, item.Label, statusLabel, item.Runtime.Round(time.Millisecond), item.Iterations, + item.InputTokens, item.OutputTokens, item.Result, )) } sb.WriteString("---\n\n") - sb.WriteString(buildReplyInstruction(roster)) + sb.WriteString(BuildReplyInstruction(roster)) return sb.String() } // buildReplyInstruction generates the instruction block for the parent LLM, // including a deterministic roster of all subagent tasks with their statuses. -func buildReplyInstruction(roster SubagentRoster) string { +func BuildReplyInstruction(roster SubagentRoster) string { // Count running tasks from roster running := 0 for _, e := range roster.Entries { diff --git a/internal/tools/subagent_config.go b/internal/tools/subagent_config.go index 3e209beb..1466114b 100644 --- a/internal/tools/subagent_config.go +++ b/internal/tools/subagent_config.go @@ -10,6 +10,7 @@ func DefaultSubagentConfig() SubagentConfig { MaxSpawnDepth: 1, // TS: maxSpawnDepth ?? 1 MaxChildrenPerAgent: 5, // TS: maxChildrenPerAgent ?? 5 ArchiveAfterMinutes: 60, // TS: archiveAfterMinutes ?? 60 + MaxRetries: 2, } } diff --git a/internal/tools/subagent_control.go b/internal/tools/subagent_control.go index ff0e47c2..9a59c049 100644 --- a/internal/tools/subagent_control.go +++ b/internal/tools/subagent_control.go @@ -159,6 +159,37 @@ func (sm *SubagentManager) Steer( return fmt.Sprintf("Steered subagent %q → new task spawned. %s", taskID, msg), nil } +// WaitForChildren blocks until all running tasks for parentID complete or timeout. +func (sm *SubagentManager) WaitForChildren(ctx context.Context, parentID string, timeoutSec int) ([]*SubagentTask, error) { + if timeoutSec <= 0 { + timeoutSec = 300 + } + deadline := time.After(time.Duration(timeoutSec) * time.Second) + ticker := time.NewTicker(500 * time.Millisecond) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return sm.ListTasks(parentID), ctx.Err() + case <-deadline: + return sm.ListTasks(parentID), fmt.Errorf("timeout after %ds waiting for children", timeoutSec) + case <-ticker.C: + tasks := sm.ListTasks(parentID) + allDone := true + for _, t := range tasks { + if t.Status == TaskStatusRunning { + allDone = false + break + } + } + if allDone { + return tasks, nil + } + } + } +} + func generateSubagentID() string { b := make([]byte, 8) rand.Read(b) diff --git a/internal/tools/subagent_exec.go b/internal/tools/subagent_exec.go index dbc8c6f9..b1e798b9 100644 --- a/internal/tools/subagent_exec.go +++ b/internal/tools/subagent_exec.go @@ -25,13 +25,15 @@ func (sm *SubagentManager) runTask(ctx context.Context, task *SubagentTask, call elapsed := time.Since(time.UnixMilli(task.CreatedAt)) item := AnnounceQueueItem{ - SubagentID: task.ID, - Label: task.Label, - Status: task.Status, - Result: task.Result, - Media: task.Media, - Runtime: elapsed, - Iterations: iterations, + SubagentID: task.ID, + Label: task.Label, + Status: task.Status, + Result: task.Result, + Media: task.Media, + Runtime: elapsed, + Iterations: iterations, + InputTokens: task.TotalInputTokens, + OutputTokens: task.TotalOutputTokens, } meta := AnnounceMetadata{ OriginChannel: task.OriginChannel, @@ -56,13 +58,19 @@ func (sm *SubagentManager) runTask(ctx context.Context, task *SubagentTask, call announceContent := FormatBatchedAnnounce([]AnnounceQueueItem{item}, roster) announceMeta := map[string]string{ - MetaOriginChannel: task.OriginChannel, - MetaOriginPeerKind: task.OriginPeerKind, - MetaParentAgent: task.ParentID, - "subagent_id": task.ID, - MetaSubagentLabel: task.Label, - MetaOriginTraceID: task.OriginTraceID.String(), - MetaOriginRootSpanID: task.OriginRootSpanID.String(), + MetaOriginChannel: task.OriginChannel, + MetaOriginPeerKind: task.OriginPeerKind, + MetaParentAgent: task.ParentID, + "subagent_id": task.ID, + MetaSubagentLabel: task.Label, + MetaSubagentStatus: task.Status, + MetaSubagentResult: task.Result, + MetaSubagentRuntime: fmt.Sprintf("%d", elapsed.Milliseconds()), + MetaSubagentIterations: fmt.Sprintf("%d", iterations), + MetaSubagentInputToks: fmt.Sprintf("%d", task.TotalInputTokens), + MetaSubagentOutputToks: fmt.Sprintf("%d", task.TotalOutputTokens), + MetaOriginTraceID: task.OriginTraceID.String(), + MetaOriginRootSpanID: task.OriginRootSpanID.String(), } if task.OriginLocalKey != "" { announceMeta[MetaOriginLocalKey] = task.OriginLocalKey @@ -214,15 +222,48 @@ func (sm *SubagentManager) executeTask(ctx context.Context, task *SubagentTask) llmStart := time.Now().UTC() llmSpanID := sm.emitLLMSpanStart(subTraceCtx, llmStart, iteration, model, activeProvider.Name(), messages) - resp, err := activeProvider.Chat(ctx, chatReq) + + maxRetries := task.spawnConfig.MaxRetries + if maxRetries <= 0 { + maxRetries = 2 + } + var resp *providers.ChatResponse + var err error + for attempt := 0; attempt <= maxRetries; attempt++ { + if attempt > 0 { + backoff := time.Duration(attempt) * 2 * time.Second + select { + case <-ctx.Done(): + case <-time.After(backoff): + } + if ctx.Err() != nil { + break + } + slog.Info("subagent LLM retry", "id", task.ID, "iteration", iteration, "attempt", attempt+1) + } + resp, err = activeProvider.Chat(ctx, chatReq) + if err == nil { + break + } + } + sm.emitLLMSpanEnd(subTraceCtx, llmSpanID, llmStart, resp, err) + // Accumulate token usage for cost tracking. + if resp != nil && resp.Usage != nil { + sm.mu.Lock() + task.TotalInputTokens += int64(resp.Usage.PromptTokens) + task.TotalOutputTokens += int64(resp.Usage.CompletionTokens) + sm.mu.Unlock() + } + if err != nil { sm.mu.Lock() task.Status = TaskStatusFailed task.Result = fmt.Sprintf("LLM error at iteration %d: %v", iteration, err) sm.mu.Unlock() slog.Warn("subagent LLM error", "id", task.ID, "iteration", iteration, "error", err) + go sm.persistStatus(ctx, task, iteration) return iteration } @@ -282,5 +323,9 @@ func (sm *SubagentManager) executeTask(ctx context.Context, task *SubagentTask) sm.mu.Unlock() slog.Info("subagent completed", "id", task.ID, "iterations", iteration) + + // Persist final status to DB (fire-and-forget). + go sm.persistStatus(ctx, task, iteration) + return iteration } diff --git a/internal/tools/subagent_persist.go b/internal/tools/subagent_persist.go new file mode 100644 index 00000000..c50008da --- /dev/null +++ b/internal/tools/subagent_persist.go @@ -0,0 +1,97 @@ +package tools + +import ( + "context" + "log/slog" + + "github.com/google/uuid" + + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +// detachedCtx creates a context that won't be cancelled but preserves tenant ID. +// Used for fire-and-forget DB writes that must succeed even after the parent ctx is cancelled. +func detachedCtx(ctx context.Context) context.Context { + bg := context.Background() + if tid := store.TenantIDFromContext(ctx); tid != uuid.Nil { + bg = store.WithTenantID(bg, tid) + } + return bg +} + +// persistCreate writes a new subagent task to the DB (fire-and-forget). +func (sm *SubagentManager) persistCreate(ctx context.Context, task *SubagentTask) { + if sm.taskStore == nil { + return + } + + dbCtx := detachedCtx(ctx) + + var sessionKey *string + if task.OriginSessionKey != "" { + s := task.OriginSessionKey + sessionKey = &s + } + var model, provider, originChannel, originChatID, originPeerKind, originUserID *string + if task.Model != "" { + model = &task.Model + } + if p := ParentProviderFromCtx(ctx); p != "" { + provider = &p + } + if task.OriginChannel != "" { + originChannel = &task.OriginChannel + } + if task.OriginChatID != "" { + originChatID = &task.OriginChatID + } + if task.OriginPeerKind != "" { + originPeerKind = &task.OriginPeerKind + } + if task.OriginUserID != "" { + originUserID = &task.OriginUserID + } + + data := &store.SubagentTaskData{ + BaseModel: store.BaseModel{ID: task.dbID}, + TenantID: task.OriginTenantID, + ParentAgentKey: task.ParentID, + SessionKey: sessionKey, + Subject: task.Label, + Description: task.Task, + Status: task.Status, + Depth: task.Depth, + Model: model, + Provider: provider, + OriginChannel: originChannel, + OriginChatID: originChatID, + OriginPeerKind: originPeerKind, + OriginUserID: originUserID, + } + + if err := sm.taskStore.Create(dbCtx, data); err != nil { + slog.Warn("subagent_persist: create failed", "id", task.ID, "error", err) + } +} + +// persistStatus updates status, result, iterations, and token counts in the DB (fire-and-forget). +func (sm *SubagentManager) persistStatus(ctx context.Context, task *SubagentTask, iterations int) { + if sm.taskStore == nil || task.dbID == uuid.Nil { + return + } + + dbCtx := detachedCtx(ctx) + + var result *string + if task.Result != "" { + result = &task.Result + } + + if err := sm.taskStore.UpdateStatus( + dbCtx, task.dbID, + task.Status, result, iterations, + task.TotalInputTokens, task.TotalOutputTokens, + ); err != nil { + slog.Warn("subagent_persist: update status failed", "id", task.ID, "error", err) + } +} diff --git a/internal/tools/subagent_spawn.go b/internal/tools/subagent_spawn.go new file mode 100644 index 00000000..b2f0bcc5 --- /dev/null +++ b/internal/tools/subagent_spawn.go @@ -0,0 +1,190 @@ +package tools + +import ( + "context" + "fmt" + "log/slog" + "time" + + "github.com/nextlevelbuilder/goclaw/internal/edition" + "github.com/nextlevelbuilder/goclaw/internal/store" + "github.com/nextlevelbuilder/goclaw/internal/tracing" +) + +// Spawn creates a new subagent task that runs asynchronously. +// Returns immediately with a status message. The subagent runs in a goroutine. +// modelOverride optionally overrides the LLM model for this subagent (matching TS sessions-spawn-tool.ts). +func (sm *SubagentManager) Spawn( + ctx context.Context, + parentID string, + depth int, + task, label, modelOverride string, + channel, chatID, peerKind string, + callback AsyncCallback, +) (string, error) { + cfg := sm.effectiveConfig(ctx) + + // Apply edition ceilings (Lite edition enforces lower limits). + ed := edition.Current() + if ed.MaxSubagentConcurrent > 0 && cfg.MaxConcurrent > ed.MaxSubagentConcurrent { + cfg.MaxConcurrent = ed.MaxSubagentConcurrent + } + if ed.MaxSubagentDepth > 0 && cfg.MaxSpawnDepth > ed.MaxSubagentDepth { + cfg.MaxSpawnDepth = ed.MaxSubagentDepth + } + + sm.mu.Lock() + + // Check depth limit + if depth >= cfg.MaxSpawnDepth { + sm.mu.Unlock() + return "", fmt.Errorf("spawn depth limit reached (%d/%d)", depth, cfg.MaxSpawnDepth) + } + + // Check concurrent limit (scoped per tenant for isolation). + tenantID := store.TenantIDFromContext(ctx) + running := 0 + for _, t := range sm.tasks { + if t.Status == TaskStatusRunning && t.OriginTenantID == tenantID { + running++ + } + } + if running >= cfg.MaxConcurrent { + sm.mu.Unlock() + return "", fmt.Errorf("max concurrent subagents reached (%d/%d)", running, cfg.MaxConcurrent) + } + + // Check per-parent children limit + childCount := 0 + for _, t := range sm.tasks { + if t.ParentID == parentID { + childCount++ + } + } + if childCount >= cfg.MaxChildrenPerAgent { + sm.mu.Unlock() + return "", fmt.Errorf("max children per agent reached (%d/%d)", childCount, cfg.MaxChildrenPerAgent) + } + + id := generateSubagentID() + if label == "" { + label = truncate(task, 50) + } + + subTask := &SubagentTask{ + ID: id, + ParentID: parentID, + Task: task, + Label: label, + Status: "running", + Depth: depth + 1, + Model: modelOverride, + OriginChannel: channel, + OriginChatID: chatID, + OriginPeerKind: peerKind, + OriginLocalKey: ToolLocalKeyFromCtx(ctx), + OriginUserID: store.UserIDFromContext(ctx), + OriginSessionKey: ToolSessionKeyFromCtx(ctx), + OriginTenantID: store.TenantIDFromContext(ctx), + OriginTraceID: tracing.TraceIDFromContext(ctx), + OriginRootSpanID: tracing.ParentSpanIDFromContext(ctx), + CreatedAt: time.Now().UnixMilli(), + spawnConfig: cfg, + } + // Detach from parent's cancellation chain so subagent survives after parent run completes. + // WithoutCancel preserves all context values (agent ID, workspace, trace info, etc.) + // but parent Done() no longer propagates. Manual cancel via taskCancel() still works. + detached := context.WithoutCancel(ctx) + taskCtx, taskCancel := context.WithCancel(detached) + subTask.cancelFunc = taskCancel + + // Assign DB UUID inside lock to avoid race with runTask goroutine. + if sm.taskStore != nil { + subTask.dbID = store.GenNewID() + } + + sm.tasks[id] = subTask + sm.mu.Unlock() + + slog.Info("subagent spawned", "id", id, "parent", parentID, "depth", subTask.Depth, "label", label) + + // Persist to DB (fire-and-forget). + if sm.taskStore != nil { + go sm.persistCreate(taskCtx, subTask) + } + + go sm.runTask(taskCtx, subTask, callback) + + return fmt.Sprintf("Spawned subagent '%s' (id=%s, depth=%d) for task: %s", + label, id, subTask.Depth, truncate(task, 100)), nil +} + +// RunSync executes a subagent task synchronously, blocking until completion. +func (sm *SubagentManager) RunSync( + ctx context.Context, + parentID string, + depth int, + task, label string, + channel, chatID string, +) (string, int, error) { + cfg := sm.effectiveConfig(ctx) + + // Apply edition ceilings (same as Spawn). + ed := edition.Current() + if ed.MaxSubagentConcurrent > 0 && cfg.MaxConcurrent > ed.MaxSubagentConcurrent { + cfg.MaxConcurrent = ed.MaxSubagentConcurrent + } + if ed.MaxSubagentDepth > 0 && cfg.MaxSpawnDepth > ed.MaxSubagentDepth { + cfg.MaxSpawnDepth = ed.MaxSubagentDepth + } + + sm.mu.Lock() + + if depth >= cfg.MaxSpawnDepth { + sm.mu.Unlock() + return "", 0, fmt.Errorf("spawn depth limit reached (%d/%d)", depth, cfg.MaxSpawnDepth) + } + + id := generateSubagentID() + if label == "" { + label = truncate(task, 50) + } + + subTask := &SubagentTask{ + ID: id, + ParentID: parentID, + Task: task, + Label: label, + Status: "running", + Depth: depth + 1, + OriginChannel: channel, + OriginChatID: chatID, + OriginLocalKey: ToolLocalKeyFromCtx(ctx), + OriginUserID: store.UserIDFromContext(ctx), + OriginSessionKey: ToolSessionKeyFromCtx(ctx), + OriginTenantID: store.TenantIDFromContext(ctx), + OriginTraceID: tracing.TraceIDFromContext(ctx), + OriginRootSpanID: tracing.ParentSpanIDFromContext(ctx), + CreatedAt: time.Now().UnixMilli(), + spawnConfig: cfg, + } + if sm.taskStore != nil { + subTask.dbID = store.GenNewID() + } + sm.tasks[id] = subTask + sm.mu.Unlock() + + slog.Info("subagent sync started", "id", id, "parent", parentID, "depth", subTask.Depth, "label", label) + + if sm.taskStore != nil { + sm.persistCreate(ctx, subTask) + } + + iterations := sm.executeTask(ctx, subTask) + + if subTask.Status == TaskStatusFailed { + return subTask.Result, iterations, fmt.Errorf("subagent failed: %s", subTask.Result) + } + + return subTask.Result, iterations, nil +} diff --git a/internal/tools/subagent_spawn_tool.go b/internal/tools/subagent_spawn_tool.go index 5a776399..c2bac9db 100644 --- a/internal/tools/subagent_spawn_tool.go +++ b/internal/tools/subagent_spawn_tool.go @@ -3,8 +3,6 @@ package tools import ( "context" "fmt" - "strings" - "time" "github.com/google/uuid" "github.com/nextlevelbuilder/goclaw/internal/store" @@ -41,7 +39,7 @@ func (t *SpawnTool) Parameters() map[string]any { "properties": map[string]any{ "action": map[string]any{ "type": "string", - "description": "'spawn' (default), 'list', 'cancel', or 'steer'", + "description": "'spawn' (default), 'list', 'cancel', 'steer', or 'wait'", }, "task": map[string]any{ "type": "string", @@ -67,6 +65,10 @@ func (t *SpawnTool) Parameters() map[string]any { "type": "string", "description": "New instructions (required for action=steer)", }, + "timeout": map[string]any{ + "type": "integer", + "description": "Timeout in seconds for action=wait (default 300)", + }, }, "required": []string{"task"}, } @@ -85,6 +87,8 @@ func (t *SpawnTool) Execute(ctx context.Context, args map[string]any) *Result { return t.executeCancel(ctx, args) case "steer": return t.executeSteer(ctx, args) + case "wait": + return t.executeWait(ctx, args) default: return t.executeSpawn(ctx, args) } @@ -180,72 +184,6 @@ func (t *SpawnTool) executeSubagentSync(ctx context.Context, args map[string]any return &Result{ForLLM: forLLM, ForUser: forUser} } -// executeList shows active subagent tasks. -func (t *SpawnTool) executeList(ctx context.Context) *Result { - parentID := ToolAgentKeyFromCtx(ctx) - if parentID == "" { - parentID = t.parentID - } - tasks := t.subagentMgr.ListTasks(parentID) - if len(tasks) == 0 { - return &Result{ForLLM: "No active tasks found."} - } - - var lines []string - running, completed, cancelled := 0, 0, 0 - for _, task := range tasks { - switch task.Status { - case "running": - running++ - case "completed": - completed++ - case "cancelled": - cancelled++ - } - line := fmt.Sprintf("- [%s] %s (id=%s, status=%s)", task.Label, truncate(task.Task, 60), task.ID, task.Status) - if task.CompletedAt > 0 { - dur := time.Duration(task.CompletedAt-task.CreatedAt) * time.Millisecond - line += fmt.Sprintf(", took %s", dur.Round(time.Millisecond)) - } - lines = append(lines, line) - } - - return &Result{ForLLM: fmt.Sprintf("Subagent tasks: %d running, %d completed, %d cancelled\n%s", - running, completed, cancelled, strings.Join(lines, "\n"))} -} - -// executeCancel cancels a subagent task by ID. -func (t *SpawnTool) executeCancel(ctx context.Context, args map[string]any) *Result { - id, _ := args["id"].(string) - if id == "" { - return ErrorResult("id is required for action=cancel") - } - - if t.subagentMgr.CancelTask(id) { - return &Result{ForLLM: fmt.Sprintf("Task '%s' cancelled.", id)} - } - - return ErrorResult(fmt.Sprintf("Task '%s' not found or not running.", id)) -} - -// executeSteer redirects a running subagent with new instructions. -func (t *SpawnTool) executeSteer(ctx context.Context, args map[string]any) *Result { - id, _ := args["id"].(string) - if id == "" { - return ErrorResult("id is required for action=steer") - } - message, _ := args["message"].(string) - if message == "" { - return ErrorResult("message is required for action=steer") - } - - msg, err := t.subagentMgr.Steer(ctx, id, message, nil) - if err != nil { - return ErrorResult(err.Error()) - } - return &Result{ForLLM: msg} -} - // SetContext is a no-op; channel/chatID are now read from ctx (thread-safe). func (t *SpawnTool) SetContext(channel, chatID string) {} @@ -254,38 +192,3 @@ func (t *SpawnTool) SetPeerKind(peerKind string) {} // SetCallback is a no-op; callback is now read from ctx (thread-safe). func (t *SpawnTool) SetCallback(cb AsyncCallback) {} - -// --- Helpers moved from old subagent_tool.go --- - -// FilterDenyList returns tool names from the registry excluding denied tools. -func FilterDenyList(reg *Registry, denyList []string) []string { - deny := make(map[string]bool, len(denyList)) - for _, n := range denyList { - deny[n] = true - } - - var allowed []string - for _, name := range reg.List() { - if !deny[name] { - allowed = append(allowed, name) - } - } - return allowed -} - -// IsSubagentDenied checks if a tool name is in the subagent deny list. -func IsSubagentDenied(toolName string, depth, maxDepth int) bool { - for _, d := range SubagentDenyAlways { - if strings.EqualFold(toolName, d) { - return true - } - } - if depth >= maxDepth { - for _, d := range SubagentDenyLeaf { - if strings.EqualFold(toolName, d) { - return true - } - } - } - return false -} diff --git a/internal/tools/subagent_spawn_tool_actions.go b/internal/tools/subagent_spawn_tool_actions.go new file mode 100644 index 00000000..b618a658 --- /dev/null +++ b/internal/tools/subagent_spawn_tool_actions.go @@ -0,0 +1,156 @@ +package tools + +import ( + "context" + "fmt" + "strings" + "time" +) + +// executeList shows active subagent tasks. +func (t *SpawnTool) executeList(ctx context.Context) *Result { + parentID := ToolAgentKeyFromCtx(ctx) + if parentID == "" { + parentID = t.parentID + } + tasks := t.subagentMgr.ListTasks(parentID) + if len(tasks) == 0 { + return &Result{ForLLM: "No active tasks found."} + } + + var lines []string + running, completed, cancelled := 0, 0, 0 + for _, task := range tasks { + switch task.Status { + case "running": + running++ + case "completed": + completed++ + case "cancelled": + cancelled++ + } + line := fmt.Sprintf("- [%s] %s (id=%s, status=%s)", task.Label, truncate(task.Task, 60), task.ID, task.Status) + if task.CompletedAt > 0 { + dur := time.Duration(task.CompletedAt-task.CreatedAt) * time.Millisecond + line += fmt.Sprintf(", took %s", dur.Round(time.Millisecond)) + } + lines = append(lines, line) + } + + return &Result{ForLLM: fmt.Sprintf("Subagent tasks: %d running, %d completed, %d cancelled\n%s", + running, completed, cancelled, strings.Join(lines, "\n"))} +} + +// executeCancel cancels a subagent task by ID. +func (t *SpawnTool) executeCancel(ctx context.Context, args map[string]any) *Result { + id, _ := args["id"].(string) + if id == "" { + return ErrorResult("id is required for action=cancel") + } + + if t.subagentMgr.CancelTask(id) { + return &Result{ForLLM: fmt.Sprintf("Task '%s' cancelled.", id)} + } + + return ErrorResult(fmt.Sprintf("Task '%s' not found or not running.", id)) +} + +// executeSteer redirects a running subagent with new instructions. +func (t *SpawnTool) executeSteer(ctx context.Context, args map[string]any) *Result { + id, _ := args["id"].(string) + if id == "" { + return ErrorResult("id is required for action=steer") + } + message, _ := args["message"].(string) + if message == "" { + return ErrorResult("message is required for action=steer") + } + + msg, err := t.subagentMgr.Steer(ctx, id, message, nil) + if err != nil { + return ErrorResult(err.Error()) + } + return &Result{ForLLM: msg} +} + +// executeWait blocks until all children of the calling agent complete or timeout. +func (t *SpawnTool) executeWait(ctx context.Context, args map[string]any) *Result { + parentID := ToolAgentKeyFromCtx(ctx) + if parentID == "" { + parentID = t.parentID + } + + timeout := 300 + if v, ok := args["timeout"].(float64); ok && v > 0 { + timeout = int(v) + } + + tasks, err := t.subagentMgr.WaitForChildren(ctx, parentID, timeout) + return t.formatWaitResult(tasks, err) +} + +func (t *SpawnTool) formatWaitResult(tasks []*SubagentTask, waitErr error) *Result { + if len(tasks) == 0 { + msg := "No subagent tasks found." + if waitErr != nil { + msg += " Error: " + waitErr.Error() + } + return &Result{ForLLM: msg} + } + + var sb strings.Builder + completed, failed := 0, 0 + var totalIn, totalOut int64 + for _, task := range tasks { + switch task.Status { + case TaskStatusCompleted: + completed++ + case TaskStatusFailed: + failed++ + } + totalIn += task.TotalInputTokens + totalOut += task.TotalOutputTokens + sb.WriteString(fmt.Sprintf("- [%s] %s: %s\n", task.Status, task.Label, truncate(task.Result, 200))) + } + + header := fmt.Sprintf("All %d subagent tasks finished (%d completed, %d failed). Total tokens: %d in / %d out\n\n", + len(tasks), completed, failed, totalIn, totalOut) + if waitErr != nil { + header = fmt.Sprintf("Wait ended: %s. %d tasks total.\n\n", waitErr.Error(), len(tasks)) + } + + return &Result{ForLLM: header + sb.String()} +} + +// FilterDenyList returns tool names from the registry excluding denied tools. +func FilterDenyList(reg *Registry, denyList []string) []string { + deny := make(map[string]bool, len(denyList)) + for _, n := range denyList { + deny[n] = true + } + + var allowed []string + for _, name := range reg.List() { + if !deny[name] { + allowed = append(allowed, name) + } + } + return allowed +} + +// IsSubagentDenied checks if a tool name is in the subagent deny list. +func IsSubagentDenied(toolName string, depth, maxDepth int) bool { + for _, d := range SubagentDenyAlways { + if strings.EqualFold(toolName, d) { + return true + } + } + if depth >= maxDepth { + for _, d := range SubagentDenyLeaf { + if strings.EqualFold(toolName, d) { + return true + } + } + } + return false +} diff --git a/internal/tools/team_metadata_keys.go b/internal/tools/team_metadata_keys.go index c12f5006..d4af0f44 100644 --- a/internal/tools/team_metadata_keys.go +++ b/internal/tools/team_metadata_keys.go @@ -20,7 +20,13 @@ const ( MetaTeamWorkspace = "team_workspace" MetaLeaderAgentID = "leader_agent_id" MetaParentAgent = "parent_agent" - MetaSubagentLabel = "subagent_label" + MetaSubagentLabel = "subagent_label" + MetaSubagentStatus = "subagent_status" + MetaSubagentResult = "subagent_result" + MetaSubagentRuntime = "subagent_runtime_ms" + MetaSubagentIterations = "subagent_iterations" + MetaSubagentInputToks = "subagent_input_tokens" + MetaSubagentOutputToks = "subagent_output_tokens" MetaCommand = "command" MetaIsForum = "is_forum" MetaMessageThreadID = "message_thread_id"