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
This commit is contained in:
viettranx committed 2026-03-31 11:45:16 +07:00
1 parent d8fc97ec63
commit 2c1ef25392
17 files changed
+913 -228

No files matched your search

+5 -2
View File
@@ -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
+2 -1
View File
@@ -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,
}
+1
View File
@@ -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
}
+45 -75
View File
@@ -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
}
+70
View File
@@ -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()
}
+202
View File
@@ -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)
}
}
+1
View File
@@ -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
}
+22 -18
View File
@@ -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 ---
+16 -12
View File
@@ -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 {
+1
View File
@@ -10,6 +10,7 @@ func DefaultSubagentConfig() SubagentConfig {
MaxSpawnDepth: 1, // TS: maxSpawnDepth ?? 1
MaxChildrenPerAgent: 5, // TS: maxChildrenPerAgent ?? 5
ArchiveAfterMinutes: 60, // TS: archiveAfterMinutes ?? 60
MaxRetries: 2,
}
}
+31
View File
@@ -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)
+60 -15
View File
@@ -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
}
+97
View File
@@ -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)
}
}
+190
View File
@@ -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
}
+7 -104
View File
@@ -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
}
@@ -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
}
+7 -1
View File
@@ -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"