feat: add async spawn manager with announcer protocol
Async spawn (fire-and-forget) with per-parent semaphore-based concurrency limiting and buffered announcement channels for result delivery. Includes: - SpawnManager: goroutine-based async agent invocation with configurable per-parent concurrency limits and timeouts - Announcer: per-session buffered channels with back-pressure (drops oldest on overflow) for spawn result delivery - SpawnTool: LLM-callable tool for async agent spawning with allowlist enforcement and capability-based routing - Config: MaxChildrenPerAgent + SpawnTimeoutSec in SubagentsConfig
This commit is contained in:
Leandro Barbosa
parent
b80b2b15f0
commit
0b7d3b9198
4 changed files with 547 additions and 2 deletions
|
|
@ -125,6 +125,8 @@ type AgentConfig struct {
|
|||
type SubagentsConfig struct {
|
||||
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 {
|
||||
|
|
|
|||
139
pkg/multiagent/announce.go
Normal file
139
pkg/multiagent/announce.go
Normal file
|
|
@ -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)
|
||||
}
|
||||
235
pkg/multiagent/spawn.go
Normal file
235
pkg/multiagent/spawn.go
Normal file
|
|
@ -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)
|
||||
}
|
||||
169
pkg/multiagent/spawn_tool.go
Normal file
169
pkg/multiagent/spawn_tool.go
Normal file
|
|
@ -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),
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Reference in a new issue