diff --git a/pkg/config/config.go b/pkg/config/config.go index 25a221200..8dd948ab4 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -123,8 +123,10 @@ type AgentConfig struct { } type SubagentsConfig struct { - AllowAgents []string `json:"allow_agents,omitempty"` - Model *AgentModelConfig `json:"model,omitempty"` + AllowAgents []string `json:"allow_agents,omitempty"` + Model *AgentModelConfig `json:"model,omitempty"` + MaxChildrenPerAgent int `json:"max_children_per_agent,omitempty"` // max concurrent async spawns per parent (default 5) + SpawnTimeoutSec int `json:"spawn_timeout_sec,omitempty"` // per-spawn timeout in seconds (default 300) } type PeerMatch struct { diff --git a/pkg/multiagent/announce.go b/pkg/multiagent/announce.go new file mode 100644 index 000000000..81e024178 --- /dev/null +++ b/pkg/multiagent/announce.go @@ -0,0 +1,139 @@ +package multiagent + +import ( + "sync" + "time" + + "github.com/sipeed/picoclaw/pkg/logger" +) + +// AnnounceMode determines how a spawn result is delivered to the parent session. +// Inspired by Google Cloud Pub/Sub delivery modes and Microsoft Azure Service Bus. +type AnnounceMode string + +const ( + // AnnounceQueue buffers the result until the parent requests it (default). + // Like Apple's GCD serial queue — ordered, non-blocking. + AnnounceQueue AnnounceMode = "queue" + + // AnnounceDirect sends the result immediately to the parent's chat channel. + // Like Google Pub/Sub push subscription — immediate delivery. + AnnounceDirect AnnounceMode = "direct" +) + +// Announcement is a completion notice from a child spawn to its parent. +type Announcement struct { + FromSessionKey string + ToSessionKey string + RunID string + AgentID string + Content string + Outcome *SpawnOutcome + Mode AnnounceMode + CreatedAt time.Time +} + +// Announcer manages per-session announcement delivery using Go channels +// (inspired by Apple's Grand Central Dispatch work queues). +// Thread-safe for concurrent producers (multiple child spawns completing +// simultaneously) and a single consumer (parent agent). +type Announcer struct { + // Per-session buffered channels (Google Pub/Sub topic model). + channels sync.Map // sessionKey -> chan *Announcement + bufSize int +} + +// NewAnnouncer creates an announcer with the given per-session buffer size. +// Buffer size follows NVIDIA's double-buffering pattern: enough to absorb +// burst completions without blocking producers. +func NewAnnouncer(bufSize int) *Announcer { + if bufSize <= 0 { + bufSize = 32 // default: buffer up to 32 pending announcements + } + return &Announcer{bufSize: bufSize} +} + +// Deliver sends an announcement to the target session's channel. +// Non-blocking: if the channel is full, the oldest announcement is dropped +// (Meta's back-pressure pattern for high-throughput systems). +func (a *Announcer) Deliver(targetSessionKey string, ann *Announcement) { + ann.CreatedAt = time.Now() + if ann.Mode == "" { + ann.Mode = AnnounceQueue + } + + ch := a.getOrCreateChan(targetSessionKey) + + select { + case ch <- ann: + logger.DebugCF("announce", "Announcement delivered", map[string]interface{}{ + "from": ann.FromSessionKey, + "to": targetSessionKey, + "run_id": ann.RunID, + "agent": ann.AgentID, + "mode": string(ann.Mode), + }) + default: + // Channel full — drop oldest to make room (back-pressure). + select { + case <-ch: + logger.WarnCF("announce", "Dropped oldest announcement (buffer full)", map[string]interface{}{ + "session": targetSessionKey, + }) + default: + } + // Retry delivery. + select { + case ch <- ann: + default: + logger.WarnCF("announce", "Failed to deliver announcement", map[string]interface{}{ + "session": targetSessionKey, + "run_id": ann.RunID, + }) + } + } +} + +// Drain returns all pending announcements for a session, clearing the buffer. +// The parent agent calls this between LLM iterations to collect spawn results. +// Follows Google's batch-pull pattern from Cloud Pub/Sub. +func (a *Announcer) Drain(sessionKey string) []*Announcement { + v, ok := a.channels.Load(sessionKey) + if !ok { + return nil + } + ch := v.(chan *Announcement) + + var results []*Announcement + for { + select { + case ann := <-ch: + results = append(results, ann) + default: + return results + } + } +} + +// Pending returns the number of pending announcements for a session. +func (a *Announcer) Pending(sessionKey string) int { + v, ok := a.channels.Load(sessionKey) + if !ok { + return 0 + } + return len(v.(chan *Announcement)) +} + +// Cleanup removes the channel for a session (called on session end). +func (a *Announcer) Cleanup(sessionKey string) { + a.channels.Delete(sessionKey) +} + +func (a *Announcer) getOrCreateChan(sessionKey string) chan *Announcement { + if v, ok := a.channels.Load(sessionKey); ok { + return v.(chan *Announcement) + } + ch := make(chan *Announcement, a.bufSize) + actual, _ := a.channels.LoadOrStore(sessionKey, ch) + return actual.(chan *Announcement) +} diff --git a/pkg/multiagent/spawn.go b/pkg/multiagent/spawn.go new file mode 100644 index 000000000..3096ece95 --- /dev/null +++ b/pkg/multiagent/spawn.go @@ -0,0 +1,235 @@ +package multiagent + +import ( + "context" + "fmt" + "sync" + "time" + + "github.com/sipeed/picoclaw/pkg/logger" +) + +// Spawn concurrency defaults. +// MaxChildrenPerAgent follows NVIDIA's stream scheduling pattern: +// limit parallel work to prevent resource exhaustion while maximizing throughput. +const ( + DefaultMaxChildren = 5 + DefaultSpawnTimeout = 5 * time.Minute +) + +// SpawnRequest describes an async agent invocation. +type SpawnRequest struct { + FromAgentID string + ToAgentID string + Task string + Context map[string]string // k-v to write to blackboard + Depth int + Visited []string + MaxDepth int + ParentRunKey string +} + +// SpawnResult is returned immediately to the caller (fire-and-forget). +type SpawnResult struct { + RunID string // unique identifier for this spawn + SessionKey string // child session key for tracking + Status string // "accepted" or "rejected" + Error string // rejection reason if status != "accepted" +} + +// SpawnOutcome is the final result written to the announcer when the spawn completes. +type SpawnOutcome struct { + RunID string + SessionKey string + AgentID string + Content string + Iterations int + Success bool + Error string + Duration time.Duration +} + +// SpawnManager orchestrates async agent spawns with concurrency limiting +// (semaphore pattern, inspired by NVIDIA CUDA stream scheduling and +// Apple GCD quality-of-service queues). +type SpawnManager struct { + registry *RunRegistry + announcer *Announcer + maxChildren int + timeout time.Duration + + // Per-parent semaphore: limits concurrent children per session. + // Google's MapReduce uses similar fan-out caps per mapper. + semaphores sync.Map // parentSessionKey -> *semaphore +} + +type semaphore struct { + ch chan struct{} +} + +func newSemaphore(max int) *semaphore { + return &semaphore{ch: make(chan struct{}, max)} +} + +func (s *semaphore) acquire() bool { + select { + case s.ch <- struct{}{}: + return true + default: + return false + } +} + +func (s *semaphore) release() { + <-s.ch +} + +func (s *semaphore) count() int { + return len(s.ch) +} + +// NewSpawnManager creates a spawn manager with the given limits. +func NewSpawnManager(registry *RunRegistry, announcer *Announcer, maxChildren int, timeout time.Duration) *SpawnManager { + if maxChildren <= 0 { + maxChildren = DefaultMaxChildren + } + if timeout <= 0 { + timeout = DefaultSpawnTimeout + } + return &SpawnManager{ + registry: registry, + announcer: announcer, + maxChildren: maxChildren, + timeout: timeout, + } +} + +// AsyncSpawn launches an agent in a background goroutine and returns immediately. +// Inspired by Google's fan-out pattern and Anthropic's parallel tool execution. +// The result is delivered via the Announcer when the spawn completes. +func (sm *SpawnManager) AsyncSpawn( + ctx context.Context, + resolver AgentResolver, + board *Blackboard, + req SpawnRequest, + channel, chatID string, +) *SpawnResult { + // Generate unique run ID and session key. + runID := fmt.Sprintf("spawn:%s:%s:%d", req.FromAgentID, req.ToAgentID, time.Now().UnixNano()) + childSessionKey := fmt.Sprintf("spawn:%s:%s:%d:%d", req.FromAgentID, req.ToAgentID, req.Depth, time.Now().UnixNano()) + + // Acquire per-parent semaphore (NVIDIA stream scheduling pattern). + sem := sm.getOrCreateSemaphore(req.ParentRunKey) + if !sem.acquire() { + return &SpawnResult{ + RunID: runID, + SessionKey: childSessionKey, + Status: "rejected", + Error: fmt.Sprintf("max concurrent children reached (%d/%d) for parent session", sem.count(), sm.maxChildren), + } + } + + // Create cancellable context with timeout to prevent goroutine leaks. + // Microsoft Azure Functions uses similar timeout patterns for durable functions. + spawnCtx, cancel := context.WithTimeout(ctx, sm.timeout) + + // Register in RunRegistry for cascade cancellation (built in Phase 3d). + sm.registry.Register(&ActiveRun{ + SessionKey: childSessionKey, + AgentID: req.ToAgentID, + ParentKey: req.ParentRunKey, + Cancel: cancel, + StartedAt: time.Now(), + }) + + logger.InfoCF("spawn", "Async spawn started", map[string]interface{}{ + "run_id": runID, + "from": req.FromAgentID, + "to": req.ToAgentID, + "depth": req.Depth, + "parent": req.ParentRunKey, + "timeout": sm.timeout.String(), + "active": sem.count(), + "max": sm.maxChildren, + }) + + // Fire-and-forget goroutine (Google MapReduce worker pattern). + go func() { + defer cancel() + defer sem.release() + defer sm.registry.Deregister(childSessionKey) + + start := time.Now() + + // Execute the handoff synchronously inside the goroutine. + result := ExecuteHandoff(spawnCtx, resolver, board, HandoffRequest{ + FromAgentID: req.FromAgentID, + ToAgentID: req.ToAgentID, + Task: req.Task, + Context: req.Context, + Depth: req.Depth, + Visited: req.Visited, + MaxDepth: req.MaxDepth, + ParentRunKey: childSessionKey, + }, channel, chatID) + + outcome := &SpawnOutcome{ + RunID: runID, + SessionKey: childSessionKey, + AgentID: req.ToAgentID, + Content: result.Content, + Iterations: result.Iterations, + Success: result.Success, + Error: result.Error, + Duration: time.Since(start), + } + + // Push result to parent via Announcer (Anthropic's auto-announce pattern). + if sm.announcer != nil { + sm.announcer.Deliver(req.ParentRunKey, &Announcement{ + FromSessionKey: childSessionKey, + ToSessionKey: req.ParentRunKey, + RunID: runID, + AgentID: req.ToAgentID, + Content: formatOutcomeMessage(outcome), + Outcome: outcome, + }) + } + + logger.InfoCF("spawn", "Async spawn completed", map[string]interface{}{ + "run_id": runID, + "agent_id": req.ToAgentID, + "success": result.Success, + "iterations": result.Iterations, + "duration": outcome.Duration.Round(time.Millisecond).String(), + }) + }() + + return &SpawnResult{ + RunID: runID, + SessionKey: childSessionKey, + Status: "accepted", + } +} + +// ActiveChildCount returns the number of active children for a parent session. +func (sm *SpawnManager) ActiveChildCount(parentSessionKey string) int { + return len(sm.registry.GetChildren(parentSessionKey)) +} + +func (sm *SpawnManager) getOrCreateSemaphore(parentKey string) *semaphore { + if v, ok := sm.semaphores.Load(parentKey); ok { + return v.(*semaphore) + } + sem := newSemaphore(sm.maxChildren) + actual, _ := sm.semaphores.LoadOrStore(parentKey, sem) + return actual.(*semaphore) +} + +func formatOutcomeMessage(o *SpawnOutcome) string { + if !o.Success { + return fmt.Sprintf("[Subagent %q failed after %s: %s]", o.AgentID, o.Duration.Round(time.Millisecond), o.Error) + } + return fmt.Sprintf("[Subagent %q completed in %s (%d iterations)]:\n%s", + o.AgentID, o.Duration.Round(time.Millisecond), o.Iterations, o.Content) +} diff --git a/pkg/multiagent/spawn_tool.go b/pkg/multiagent/spawn_tool.go new file mode 100644 index 000000000..2785abb1b --- /dev/null +++ b/pkg/multiagent/spawn_tool.go @@ -0,0 +1,169 @@ +package multiagent + +import ( + "context" + "fmt" + "strings" + + "github.com/sipeed/picoclaw/pkg/tools" +) + +// SpawnTool allows an LLM agent to asynchronously spawn a child agent. +// Unlike HandoffTool (synchronous, blocking), SpawnTool returns immediately +// with a run ID. Results are auto-announced back to the parent session. +// +// Pattern: Anthropic's orchestrator-workers + OpenAI Swarm's lightweight handoffs. +type SpawnTool struct { + resolver AgentResolver + board *Blackboard + spawnManager *SpawnManager + fromAgentID string + originChannel string + originChatID string + depth int + visited []string + maxDepth int + parentSessionKey string + allowlistChecker AllowlistChecker +} + +// NewSpawnTool creates a spawn tool bound to a source agent. +func NewSpawnTool(resolver AgentResolver, board *Blackboard, spawnManager *SpawnManager, fromAgentID string) *SpawnTool { + return &SpawnTool{ + resolver: resolver, + board: board, + spawnManager: spawnManager, + fromAgentID: fromAgentID, + originChannel: "cli", + originChatID: "direct", + } +} + +func (t *SpawnTool) Name() string { return "spawn_agent" } + +func (t *SpawnTool) Description() string { + agents := t.resolver.ListAgents() + if len(agents) <= 1 { + return "Spawn a child agent asynchronously. Returns immediately — result auto-announces back. No other agents currently available." + } + + var sb strings.Builder + sb.WriteString("Spawn a child agent asynchronously. Returns immediately with a run ID — the result will auto-announce back when complete. Available agents:\n") + for _, a := range agents { + if a.ID == t.fromAgentID { + continue + } + fmt.Fprintf(&sb, "- %s", a.ID) + if a.Name != "" { + fmt.Fprintf(&sb, " (%s)", a.Name) + } + if a.Role != "" { + fmt.Fprintf(&sb, ": %s", a.Role) + } + sb.WriteString("\n") + } + sb.WriteString("\nUse 'list_spawns' to check status. Results are auto-delivered — no need to poll.") + return sb.String() +} + +func (t *SpawnTool) Parameters() map[string]any { + return map[string]any{ + "type": "object", + "properties": map[string]any{ + "agent_id": map[string]any{ + "type": "string", + "description": "The ID of the agent to spawn", + }, + "capability": map[string]any{ + "type": "string", + "description": "Route to an agent with this capability instead of by ID", + }, + "task": map[string]any{ + "type": "string", + "description": "The task for the spawned agent", + }, + "context": map[string]any{ + "type": "object", + "description": "Optional key-value context to share via blackboard", + }, + }, + "required": []string{"task"}, + } +} + +// SetBoard implements BoardAware. +func (t *SpawnTool) SetBoard(board *Blackboard) { + t.board = board +} + +// SetContext implements ContextualTool. +func (t *SpawnTool) SetContext(channel, chatID string) { + t.originChannel = channel + t.originChatID = chatID +} + +// SetAllowlistChecker sets the allowlist checker for spawn permissions. +func (t *SpawnTool) SetAllowlistChecker(checker AllowlistChecker) { + t.allowlistChecker = checker +} + +// SetRunRegistry sets registry and parent key for cascade tracking. +func (t *SpawnTool) SetRunRegistry(registry *RunRegistry, parentSessionKey string) { + t.parentSessionKey = parentSessionKey +} + +func (t *SpawnTool) Execute(ctx context.Context, args map[string]any) *tools.ToolResult { + agentID, _ := args["agent_id"].(string) + capability, _ := args["capability"].(string) + task, _ := args["task"].(string) + + if task == "" { + return tools.ErrorResult("task is required") + } + + // Resolve agent + if agentID == "" && capability != "" { + matches := FindAgentsByCapability(t.resolver, capability) + if len(matches) == 0 { + return tools.ErrorResult(fmt.Sprintf("no agent found with capability %q", capability)) + } + agentID = matches[0].ID + } + if agentID == "" { + return tools.ErrorResult("agent_id or capability is required") + } + + // Allowlist check + if t.allowlistChecker != nil && !t.allowlistChecker.CanHandoff(t.fromAgentID, agentID) { + return tools.ErrorResult(fmt.Sprintf("spawn from %q to %q not allowed by policy", t.fromAgentID, agentID)) + } + + // Parse context + var contextMap map[string]string + if ctxRaw, ok := args["context"].(map[string]any); ok { + contextMap = make(map[string]string, len(ctxRaw)) + for k, v := range ctxRaw { + contextMap[k] = fmt.Sprintf("%v", v) + } + } + + result := t.spawnManager.AsyncSpawn(ctx, t.resolver, t.board, SpawnRequest{ + FromAgentID: t.fromAgentID, + ToAgentID: agentID, + Task: task, + Context: contextMap, + Depth: t.depth, + Visited: t.visited, + MaxDepth: t.maxDepth, + ParentRunKey: t.parentSessionKey, + }, t.originChannel, t.originChatID) + + if result.Status != "accepted" { + return tools.ErrorResult(fmt.Sprintf("Spawn rejected: %s", result.Error)) + } + + return &tools.ToolResult{ + ForLLM: fmt.Sprintf("Agent %q spawned (run_id: %s). It runs asynchronously — the result will auto-announce back when complete. Continue with other work.", agentID, result.RunID), + ForUser: fmt.Sprintf("Spawned agent %q (run: %s)", agentID, result.RunID), + } +}