feat: stabilize swarm engine and enhance observability

Summary of changes:
1. Hard-coded Limits (Fixed): Added SummarizeThreshold and connected limits to config.
2. Silent Database Failures (Fixed): Enhanced error handling and logging for SQLite and NodeActor transitions.
3. "Zombie" Agents (Fixed): Implemented StopAll/Stop methods for graceful shutdown and linked to AgentLoop.
4. Memory ID Corruption (Verified): Ensured valid float formatting and added timestamps for uniqueness.
5. Code Quality: Corrected typos in system prompts and improved /swarm command error reporting.
This commit is contained in:
Danieldd28 2026-02-11 01:05:45 +07:00
parent 52e3470e06
commit 6bdf30a777
8 changed files with 194 additions and 76 deletions

View file

@ -133,6 +133,9 @@ func (al *AgentLoop) Run(ctx context.Context) error {
func (al *AgentLoop) Stop() { func (al *AgentLoop) Stop() {
al.running = false al.running = false
if al.swarm != nil {
al.swarm.Stop()
}
} }
func (al *AgentLoop) ProcessDirect(ctx context.Context, content, sessionKey string) (string, error) { func (al *AgentLoop) ProcessDirect(ctx context.Context, content, sessionKey string) (string, error) {

View file

@ -15,12 +15,13 @@ type MemoryConfig struct {
} }
type LimitsConfig struct { type LimitsConfig struct {
MaxDepth int `json:"max_depth" env:"PICOCLAW_SWARM_MAX_DEPTH"` MaxDepth int `json:"max_depth" env:"PICOCLAW_SWARM_MAX_DEPTH"`
MaxNodes int `json:"max_nodes" env:"PICOCLAW_SWARM_MAX_NODES"` MaxNodes int `json:"max_nodes" env:"PICOCLAW_SWARM_MAX_NODES"`
GlobalTimeout time.Duration `json:"global_timeout" env:"PICOCLAW_SWARM_GLOBAL_TIMEOUT"` GlobalTimeout time.Duration `json:"global_timeout" env:"PICOCLAW_SWARM_GLOBAL_TIMEOUT"`
MaxRetries int `json:"max_retries" env:"PICOCLAW_SWARM_MAX_RETRIES"` MaxRetries int `json:"max_retries" env:"PICOCLAW_SWARM_MAX_RETRIES"`
MaxIterations int `json:"max_iterations" env:"PICOCLAW_SWARM_MAX_ITERATIONS"` MaxIterations int `json:"max_iterations" env:"PICOCLAW_SWARM_MAX_ITERATIONS"`
PruningMsgKeep int `json:"pruning_msg_keep" env:"PICOCLAW_SWARM_PRUNING_MSG_KEEP"` PruningMsgKeep int `json:"pruning_msg_keep" env:"PICOCLAW_SWARM_PRUNING_MSG_KEEP"`
SummarizeThreshold int `json:"summarize_threshold" env:"PICOCLAW_SWARM_SUMMARIZE_THRESHOLD"`
} }
type ResilienceConfig struct { type ResilienceConfig struct {
@ -44,12 +45,13 @@ type Policy struct {
func DefaultSwarmConfig() SwarmConfig { func DefaultSwarmConfig() SwarmConfig {
return SwarmConfig{ return SwarmConfig{
Limits: LimitsConfig{ Limits: LimitsConfig{
MaxDepth: 3, MaxDepth: 3,
MaxNodes: 10, MaxNodes: 10,
GlobalTimeout: 10 * time.Minute, GlobalTimeout: 10 * time.Minute,
MaxRetries: 3, MaxRetries: 3,
MaxIterations: 10, MaxIterations: 10,
PruningMsgKeep: 6, // Keep last 6 messages when pruning PruningMsgKeep: 6, // Keep last 6 messages when pruning
SummarizeThreshold: 20, // Summarize after 20 messages
}, },
Resilience: ResilienceConfig{ Resilience: ResilienceConfig{
RetryBackoff: 2 * time.Second, RetryBackoff: 2 * time.Second,

View file

@ -4,6 +4,7 @@ import (
"context" "context"
"fmt" "fmt"
"runtime" "runtime"
"time"
"github.com/philippgille/chromem-go" "github.com/philippgille/chromem-go"
"github.com/sipeed/picoclaw/pkg/swarm/core" "github.com/sipeed/picoclaw/pkg/swarm/core"
@ -42,7 +43,7 @@ func (s *ChromemStore) SaveFact(ctx context.Context, fact core.Fact) error {
meta["source"] = fact.Source meta["source"] = fact.Source
doc := chromem.Document{ doc := chromem.Document{
ID: fmt.Sprintf("%s_%d", fact.SwarmID, fact.Confidence), ID: fmt.Sprintf("%s_%f_%d", fact.SwarmID, fact.Confidence, time.Now().UnixNano()),
Content: fact.Content, Content: fact.Content,
Metadata: meta, Metadata: meta,
} }

View file

@ -17,14 +17,23 @@ func NewSQLiteStore(dbPath string) (*SQLiteStore, error) {
db, err := sql.Open("sqlite", dbPath) db, err := sql.Open("sqlite", dbPath)
if err != nil { return nil, err } if err != nil { return nil, err }
s := &SQLiteStore{db: db} s := &SQLiteStore{db: db}
s.init() if err := s.init(); err != nil {
return nil, err
}
return s, nil return s, nil
} }
func (s *SQLiteStore) init() { func (s *SQLiteStore) init() error {
s.db.Exec(`CREATE TABLE IF NOT EXISTS swarms (id TEXT PRIMARY KEY, goal TEXT, status TEXT, created_at DATETIME);`) if _, err := s.db.Exec(`CREATE TABLE IF NOT EXISTS swarms (id TEXT PRIMARY KEY, goal TEXT, status TEXT, created_at DATETIME);`); err != nil {
s.db.Exec(`CREATE TABLE IF NOT EXISTS nodes (id TEXT PRIMARY KEY, swarm_id TEXT, parent_id TEXT, role JSON, task TEXT, status TEXT, output TEXT, stats JSON);`) return err
s.db.Exec(`CREATE TABLE IF NOT EXISTS facts (swarm_id TEXT, content TEXT, confidence REAL, source TEXT, metadata JSON);`) }
if _, err := s.db.Exec(`CREATE TABLE IF NOT EXISTS nodes (id TEXT PRIMARY KEY, swarm_id TEXT, parent_id TEXT, role JSON, task TEXT, status TEXT, output TEXT, stats JSON);`); err != nil {
return err
}
if _, err := s.db.Exec(`CREATE TABLE IF NOT EXISTS facts (swarm_id TEXT, content TEXT, confidence REAL, source TEXT, metadata JSON);`); err != nil {
return err
}
return nil
} }
func (s *SQLiteStore) CreateSwarm(ctx context.Context, sw *core.Swarm) error { func (s *SQLiteStore) CreateSwarm(ctx context.Context, sw *core.Swarm) error {
@ -35,8 +44,10 @@ func (s *SQLiteStore) CreateSwarm(ctx context.Context, sw *core.Swarm) error {
func (s *SQLiteStore) GetSwarm(ctx context.Context, id core.SwarmID) (*core.Swarm, error) { func (s *SQLiteStore) GetSwarm(ctx context.Context, id core.SwarmID) (*core.Swarm, error) {
row := s.db.QueryRowContext(ctx, "SELECT id, goal, status, created_at FROM swarms WHERE id=?", id) row := s.db.QueryRowContext(ctx, "SELECT id, goal, status, created_at FROM swarms WHERE id=?", id)
var sw core.Swarm var sw core.Swarm
err := row.Scan(&sw.ID, &sw.Goal, &sw.Status, &sw.CreatedAt) if err := row.Scan(&sw.ID, &sw.Goal, &sw.Status, &sw.CreatedAt); err != nil {
return &sw, err return nil, err
}
return &sw, nil
} }
func (s *SQLiteStore) UpdateSwarm(ctx context.Context, sw *core.Swarm) error { func (s *SQLiteStore) UpdateSwarm(ctx context.Context, sw *core.Swarm) error {
@ -45,21 +56,32 @@ func (s *SQLiteStore) UpdateSwarm(ctx context.Context, sw *core.Swarm) error {
} }
func (s *SQLiteStore) ListSwarms(ctx context.Context, status core.SwarmStatus) ([]*core.Swarm, error) { func (s *SQLiteStore) ListSwarms(ctx context.Context, status core.SwarmStatus) ([]*core.Swarm, error) {
rows, _ := s.db.QueryContext(ctx, "SELECT id, goal, status, created_at FROM swarms WHERE status=?", status) rows, err := s.db.QueryContext(ctx, "SELECT id, goal, status, created_at FROM swarms WHERE status=?", status)
if err != nil {
return nil, err
}
defer rows.Close() defer rows.Close()
var list []*core.Swarm var list []*core.Swarm
for rows.Next() { for rows.Next() {
var sw core.Swarm var sw core.Swarm
rows.Scan(&sw.ID, &sw.Goal, &sw.Status, &sw.CreatedAt) if err := rows.Scan(&sw.ID, &sw.Goal, &sw.Status, &sw.CreatedAt); err != nil {
return nil, err
}
list = append(list, &sw) list = append(list, &sw)
} }
return list, nil return list, nil
} }
func (s *SQLiteStore) CreateNode(ctx context.Context, n *core.Node) error { func (s *SQLiteStore) CreateNode(ctx context.Context, n *core.Node) error {
role, _ := json.Marshal(n.Role) role, err := json.Marshal(n.Role)
stats, _ := json.Marshal(n.Stats) if err != nil {
_, err := s.db.ExecContext(ctx, "INSERT INTO nodes VALUES (?, ?, ?, ?, ?, ?, ?, ?)", n.ID, n.SwarmID, n.ParentID, role, n.Task, n.Status, n.Output, stats) return err
}
stats, err := json.Marshal(n.Stats)
if err != nil {
return err
}
_, err = s.db.ExecContext(ctx, "INSERT INTO nodes VALUES (?, ?, ?, ?, ?, ?, ?, ?)", n.ID, n.SwarmID, n.ParentID, role, n.Task, n.Status, n.Output, stats)
return err return err
} }
@ -67,36 +89,57 @@ func (s *SQLiteStore) GetNode(ctx context.Context, id core.NodeID) (*core.Node,
row := s.db.QueryRowContext(ctx, "SELECT * FROM nodes WHERE id=?", id) row := s.db.QueryRowContext(ctx, "SELECT * FROM nodes WHERE id=?", id)
var n core.Node var n core.Node
var role, stats []byte var role, stats []byte
row.Scan(&n.ID, &n.SwarmID, &n.ParentID, &role, &n.Task, &n.Status, &n.Output, &stats) if err := row.Scan(&n.ID, &n.SwarmID, &n.ParentID, &role, &n.Task, &n.Status, &n.Output, &stats); err != nil {
json.Unmarshal(role, &n.Role) return nil, err
json.Unmarshal(stats, &n.Stats) }
if err := json.Unmarshal(role, &n.Role); err != nil {
return nil, err
}
if err := json.Unmarshal(stats, &n.Stats); err != nil {
return nil, err
}
return &n, nil return &n, nil
} }
func (s *SQLiteStore) UpdateNode(ctx context.Context, n *core.Node) error { func (s *SQLiteStore) UpdateNode(ctx context.Context, n *core.Node) error {
stats, _ := json.Marshal(n.Stats) stats, err := json.Marshal(n.Stats)
_, err := s.db.ExecContext(ctx, "UPDATE nodes SET status=?, output=?, stats=? WHERE id=?", n.Status, n.Output, stats, n.ID) if err != nil {
return err
}
_, err = s.db.ExecContext(ctx, "UPDATE nodes SET status=?, output=?, stats=? WHERE id=?", n.Status, n.Output, stats, n.ID)
return err return err
} }
func (s *SQLiteStore) GetSwarmNodes(ctx context.Context, id core.SwarmID) ([]*core.Node, error) { func (s *SQLiteStore) GetSwarmNodes(ctx context.Context, id core.SwarmID) ([]*core.Node, error) {
rows, _ := s.db.QueryContext(ctx, "SELECT * FROM nodes WHERE swarm_id=?", id) rows, err := s.db.QueryContext(ctx, "SELECT * FROM nodes WHERE swarm_id=?", id)
if err != nil {
return nil, err
}
defer rows.Close() defer rows.Close()
var list []*core.Node var list []*core.Node
for rows.Next() { for rows.Next() {
var n core.Node var n core.Node
var role, stats []byte var role, stats []byte
rows.Scan(&n.ID, &n.SwarmID, &n.ParentID, &role, &n.Task, &n.Status, &n.Output, &stats) if err := rows.Scan(&n.ID, &n.SwarmID, &n.ParentID, &role, &n.Task, &n.Status, &n.Output, &stats); err != nil {
json.Unmarshal(role, &n.Role) return nil, err
json.Unmarshal(stats, &n.Stats) }
if err := json.Unmarshal(role, &n.Role); err != nil {
return nil, err
}
if err := json.Unmarshal(stats, &n.Stats); err != nil {
return nil, err
}
list = append(list, &n) list = append(list, &n)
} }
return list, nil return list, nil
} }
func (s *SQLiteStore) SaveFact(ctx context.Context, f core.Fact) error { func (s *SQLiteStore) SaveFact(ctx context.Context, f core.Fact) error {
meta, _ := json.Marshal(f.Metadata) meta, err := json.Marshal(f.Metadata)
_, err := s.db.ExecContext(ctx, "INSERT INTO facts VALUES (?, ?, ?, ?, ?)", f.SwarmID, f.Content, f.Confidence, f.Source, meta) if err != nil {
return err
}
_, err = s.db.ExecContext(ctx, "INSERT INTO facts VALUES (?, ?, ?, ?, ?)", f.SwarmID, f.Content, f.Confidence, f.Source, meta)
return err return err
} }
@ -109,12 +152,17 @@ func (s *SQLiteStore) SearchFacts(ctx context.Context, id core.SwarmID, q string
args = []any{"%" + q + "%", limit} args = []any{"%" + q + "%", limit}
} }
rows, _ := s.db.QueryContext(ctx, query, args...) rows, err := s.db.QueryContext(ctx, query, args...)
if err != nil {
return nil, err
}
defer rows.Close() defer rows.Close()
var res []core.FactResult var res []core.FactResult
for rows.Next() { for rows.Next() {
var content string var content string
rows.Scan(&content) if err := rows.Scan(&content); err != nil {
return nil, err
}
res = append(res, core.FactResult{Content: content, Score: 1.0}) res = append(res, core.FactResult{Content: content, Score: 1.0})
} }
return res, nil return res, nil

View file

@ -13,7 +13,7 @@ SWARM CONTEXT:
- Your Role: %s - Your Role: %s
INSTRUCTIONS: INSTRUCTIONS:
1. FOCUS: Stick strictly to your assigned role. Do not halllucinate capabilities you don't have. 1. FOCUS: Stick strictly to your assigned role. Do not hallucinate capabilities you don't have.
2. COLLABORATION: If you need information you can't get, ask for it clearly. 2. COLLABORATION: If you need information you can't get, ask for it clearly.
3. OUTPUT: Provide clear, structured reasoning. 3. OUTPUT: Provide clear, structured reasoning.
4. TOOLS: Use available tools to gather facts. Do not guess. 4. TOOLS: Use available tools to gather facts. Do not guess.

View file

@ -6,6 +6,7 @@ import (
"fmt" "fmt"
"log/slog" "log/slog"
"github.com/sipeed/picoclaw/pkg/swarm/config"
"github.com/sipeed/picoclaw/pkg/swarm/core" "github.com/sipeed/picoclaw/pkg/swarm/core"
"github.com/sipeed/picoclaw/pkg/tools" "github.com/sipeed/picoclaw/pkg/tools"
) )
@ -17,11 +18,12 @@ type NodeActor struct {
LLM core.LLMClient LLM core.LLMClient
Tools *tools.ToolRegistry Tools *tools.ToolRegistry
Policy *core.PolicyChecker Policy *core.PolicyChecker
Config config.SwarmConfig
peerInsights []string // Buffer for thoughts heard from peers peerInsights []string // Buffer for thoughts heard from peers
} }
func NewNodeActor(data *core.Node, bus core.EventBus, store core.SwarmStore, llm core.LLMClient, toolRegistry *tools.ToolRegistry, policy *core.PolicyChecker) *NodeActor { func NewNodeActor(data *core.Node, bus core.EventBus, store core.SwarmStore, llm core.LLMClient, toolRegistry *tools.ToolRegistry, policy *core.PolicyChecker, cfg config.SwarmConfig) *NodeActor {
return &NodeActor{ return &NodeActor{
Data: data, Data: data,
Bus: bus, Bus: bus,
@ -29,13 +31,16 @@ func NewNodeActor(data *core.Node, bus core.EventBus, store core.SwarmStore, llm
LLM: llm, LLM: llm,
Tools: toolRegistry, Tools: toolRegistry,
Policy: policy, Policy: policy,
Config: cfg,
peerInsights: make([]string, 0), peerInsights: make([]string, 0),
} }
} }
func (n *NodeActor) Start(ctx context.Context) { func (n *NodeActor) Start(ctx context.Context) {
n.Data.Status = core.NodeStatusRunning n.Data.Status = core.NodeStatusRunning
n.Store.UpdateNode(ctx, n.Data) if err := n.Store.UpdateNode(ctx, n.Data); err != nil {
slog.Error("Failed to update node status to running", "node", n.Data.ID, "error", err)
}
// Subscribe to peer events // Subscribe to peer events
sub, _ := n.Bus.Subscribe("node.events", func(e core.Event) { sub, _ := n.Bus.Subscribe("node.events", func(e core.Event) {
@ -69,35 +74,52 @@ func (n *NodeActor) run(ctx context.Context) {
allowedTools := n.getToolsForRole() allowedTools := n.getToolsForRole()
model := n.Data.Role.Model model := n.Data.Role.Model
for i := 0; i < 10; i++ { // Max 10 iterations maxIterations := n.Config.Limits.MaxIterations
// Progressive Summarization: If history > 20 messages, compress the middle part if maxIterations <= 0 {
if len(messages) > 20 { maxIterations = 10
}
summarizeThreshold := n.Config.Limits.SummarizeThreshold
if summarizeThreshold <= 0 {
summarizeThreshold = 20
}
for i := 0; i < maxIterations; i++ {
// Progressive Summarization: If history > threshold messages, compress the middle part
if len(messages) > summarizeThreshold {
slog.Info("Context threshold reached, performing progressive summarization", "node", n.Data.ID) slog.Info("Context threshold reached, performing progressive summarization", "node", n.Data.ID)
// 1. Identify context to summarize // 1. Identify context to summarize
// Header: System (0) + User Goal (1) // Header: System (0) + User Goal (1)
// Tail: Last 6 messages // Tail: Last N messages
toSummarize := messages[2 : len(messages)-6] keepCount := n.Config.Limits.PruningMsgKeep
if keepCount <= 0 {
keepCount = 6
}
summaryPrompt := "Briefly summarize the key findings, data points, and progress from the conversation above. Focus on facts discovered so far. Be very concise." if len(messages) > keepCount+2 {
summaryMessages := append(toSummarize, core.Message{Role: "user", Content: summaryPrompt}) toSummarize := messages[2 : len(messages)-keepCount]
summaryResp, err := n.LLM.Chat(ctx, summaryMessages, nil, model) summaryPrompt := "Briefly summarize the key findings, data points, and progress from the conversation above. Focus on facts discovered so far. Be very concise."
if err == nil { summaryMessages := append(toSummarize, core.Message{Role: "user", Content: summaryPrompt})
// 2. Reconstruct history: [Header] + [Summary] + [Tail]
tail := messages[len(messages)-6:]
newHistory := make([]core.Message, 0, 10)
newHistory = append(newHistory, messages[0:2]...) // Keep System + Goal
newHistory = append(newHistory, core.Message{
Role: "system",
Content: fmt.Sprintf("Previous context summary: %s", summaryResp.Content),
})
messages = append(newHistory, tail...)
n.Bus.Publish("node.events", core.Event{ summaryResp, err := n.LLM.Chat(ctx, summaryMessages, nil, model)
Type: core.EventNodeThinking, SwarmID: n.Data.SwarmID, NodeID: n.Data.ID, if err == nil {
Payload: map[string]any{"content": "🧠 I've summarized my previous findings to keep my memory sharp."}, // 2. Reconstruct history: [Header] + [Summary] + [Tail]
}) tail := messages[len(messages)-keepCount:]
newHistory := make([]core.Message, 0, keepCount+3)
newHistory = append(newHistory, messages[0:2]...) // Keep System + Goal
newHistory = append(newHistory, core.Message{
Role: "system",
Content: fmt.Sprintf("Previous context summary: %s", summaryResp.Content),
})
messages = append(newHistory, tail...)
n.Bus.Publish("node.events", core.Event{
Type: core.EventNodeThinking, SwarmID: n.Data.SwarmID, NodeID: n.Data.ID,
Payload: map[string]any{"content": "🧠 I've summarized my previous findings to keep my memory sharp."},
})
}
} }
} }
@ -177,7 +199,9 @@ func (n *NodeActor) getToolsForRole() []core.ToolDef {
func (n *NodeActor) complete(ctx context.Context, out string) { func (n *NodeActor) complete(ctx context.Context, out string) {
n.Data.Output = out n.Data.Output = out
n.Data.Status = core.NodeStatusCompleted n.Data.Status = core.NodeStatusCompleted
n.Store.UpdateNode(ctx, n.Data) if err := n.Store.UpdateNode(ctx, n.Data); err != nil {
slog.Error("Failed to update node status to completed", "node", n.Data.ID, "error", err)
}
n.Bus.Publish("node.events", core.Event{ n.Bus.Publish("node.events", core.Event{
Type: core.EventNodeCompleted, SwarmID: n.Data.SwarmID, NodeID: n.Data.ID, Type: core.EventNodeCompleted, SwarmID: n.Data.SwarmID, NodeID: n.Data.ID,
Payload: map[string]any{"output": out}, Payload: map[string]any{"output": out},
@ -186,7 +210,9 @@ func (n *NodeActor) complete(ctx context.Context, out string) {
func (n *NodeActor) fail(ctx context.Context, err error) { func (n *NodeActor) fail(ctx context.Context, err error) {
n.Data.Status = core.NodeStatusFailed n.Data.Status = core.NodeStatusFailed
n.Store.UpdateNode(ctx, n.Data) if errUpdate := n.Store.UpdateNode(ctx, n.Data); errUpdate != nil {
slog.Error("Failed to update node status to failed", "node", n.Data.ID, "error", errUpdate)
}
n.Bus.Publish("node.events", core.Event{ n.Bus.Publish("node.events", core.Event{
Type: core.EventNodeFailed, SwarmID: n.Data.SwarmID, NodeID: n.Data.ID, Type: core.EventNodeFailed, SwarmID: n.Data.SwarmID, NodeID: n.Data.ID,
Payload: map[string]any{"error": err.Error()}, Payload: map[string]any{"error": err.Error()},

View file

@ -41,14 +41,19 @@ func (o *Orchestrator) SetSharedMemory(m core.SharedMemory) { o.memory = m }
func (o *Orchestrator) SpawnSwarm(ctx context.Context, goal string) (core.SwarmID, error) { func (o *Orchestrator) SpawnSwarm(ctx context.Context, goal string) (core.SwarmID, error) {
id := core.SwarmID(uuid.New().String()) id := core.SwarmID(uuid.New().String())
o.store.CreateSwarm(ctx, &core.Swarm{ID: id, Goal: goal, Status: core.SwarmStatusActive, CreatedAt: time.Now()}) if err := o.store.CreateSwarm(ctx, &core.Swarm{ID: id, Goal: goal, Status: core.SwarmStatusActive, CreatedAt: time.Now()}); err != nil {
return "", fmt.Errorf("failed to create swarm in store: %w", err)
}
o.mu.Lock() o.mu.Lock()
sCtx, cancel := context.WithCancel(context.Background()) sCtx, cancel := context.WithCancel(ctx)
o.activeSwarms[id] = cancel o.activeSwarms[id] = cancel
o.mu.Unlock() o.mu.Unlock()
go o.RunSubTask(sCtx, id, "", "Manager", goal) go func() {
defer o.StopSwarm(id)
o.RunSubTask(sCtx, id, "", "Manager", goal)
}()
return id, nil return id, nil
} }
@ -64,7 +69,9 @@ func (o *Orchestrator) RunSubTask(ctx context.Context, sid core.SwarmID, pid cor
Role: core.Role{Name: roleName, SystemPrompt: rc.SystemPrompt, Model: model}, Role: core.Role{Name: roleName, SystemPrompt: rc.SystemPrompt, Model: model},
Task: task, Status: core.NodeStatusPending, Task: task, Status: core.NodeStatusPending,
} }
o.store.CreateNode(ctx, node) if err := o.store.CreateNode(ctx, node); err != nil {
return "", fmt.Errorf("failed to create node in store: %w", err)
}
nt := o.tools.Clone() nt := o.tools.Clone()
if roleName == "Manager" { nt.Register(NewDelegateTool(o, sid, node.ID)) } if roleName == "Manager" { nt.Register(NewDelegateTool(o, sid, node.ID)) }
@ -77,19 +84,27 @@ func (o *Orchestrator) RunSubTask(ctx context.Context, sid core.SwarmID, pid cor
fail := make(chan error, 1) fail := make(chan error, 1)
sub, _ := o.bus.Subscribe("node.events", func(e core.Event) { sub, _ := o.bus.Subscribe("node.events", func(e core.Event) {
if e.NodeID == node.ID { if e.NodeID == node.ID {
if e.Type == core.EventNodeCompleted { done <- e.Payload["output"].(string) } if e.Type == core.EventNodeCompleted {
output, _ := e.Payload["output"].(string)
done <- output
}
if e.Type == core.EventNodeFailed { fail <- fmt.Errorf("%v", e.Payload["error"]) } if e.Type == core.EventNodeFailed { fail <- fmt.Errorf("%v", e.Payload["error"]) }
} }
}) })
defer sub.Unsubscribe() defer sub.Unsubscribe()
NewNodeActor(node, o.bus, o.store, o.llm, nt, o.policyChecker).Start(ctx) NewNodeActor(node, o.bus, o.store, o.llm, nt, o.policyChecker, o.config).Start(ctx)
timeout := o.config.Limits.GlobalTimeout
if timeout <= 0 {
timeout = 15 * time.Minute
}
select { select {
case out := <-done: return out, nil case out := <-done: return out, nil
case err := <-fail: return "", err case err := <-fail: return "", err
case <-ctx.Done(): return "", ctx.Err() case <-ctx.Done(): return "", ctx.Err()
case <-time.After(15 * time.Minute): return "", core.ErrTaskTimeout case <-time.After(timeout): return "", core.ErrTaskTimeout
} }
} }
@ -101,3 +116,12 @@ func (o *Orchestrator) StopSwarm(id core.SwarmID) {
delete(o.activeSwarms, id) delete(o.activeSwarms, id)
} }
} }
func (o *Orchestrator) StopAll() {
o.mu.Lock()
defer o.mu.Unlock()
for id, cancel := range o.activeSwarms {
cancel()
delete(o.activeSwarms, id)
}
}

View file

@ -41,7 +41,7 @@ func NewService(dbPath string, provider providers.LLMProvider, registry *tools.T
} }
func (s *Service) listen() { func (s *Service) listen() {
s.Bus.Subscribe("node.events", func(e core.Event) { if _, err := s.Bus.Subscribe("node.events", func(e core.Event) {
msg := "" msg := ""
switch e.Type { switch e.Type {
case core.EventNodeThinking: msg = fmt.Sprintf("🤖 [%s]: %s", e.NodeID[:4], e.Payload["content"]) case core.EventNodeThinking: msg = fmt.Sprintf("🤖 [%s]: %s", e.NodeID[:4], e.Payload["content"])
@ -51,7 +51,9 @@ func (s *Service) listen() {
if msg != "" { if msg != "" {
select { case s.Outbound <- msg: default: } select { case s.Outbound <- msg: default: }
} }
}) }); err != nil {
fmt.Printf("Warning: Swarm event subscription failed: %v\n", err)
}
} }
func (s *Service) HandleCommand(ctx context.Context, input string) string { func (s *Service) HandleCommand(ctx context.Context, input string) string {
@ -61,10 +63,16 @@ func (s *Service) HandleCommand(ctx context.Context, input string) string {
switch args[1] { switch args[1] {
case "spawn": case "spawn":
goal := strings.Join(args[2:], " ") goal := strings.Join(args[2:], " ")
id, _ := s.Orchestrator.SpawnSwarm(ctx, goal) id, err := s.Orchestrator.SpawnSwarm(ctx, goal)
if err != nil {
return fmt.Sprintf("❌ Failed to spawn swarm: %v", err)
}
return fmt.Sprintf("🚀 Swarm ID: %s", id) return fmt.Sprintf("🚀 Swarm ID: %s", id)
case "list": case "list":
swarms, _ := s.Store.ListSwarms(ctx, core.SwarmStatusActive) swarms, err := s.Store.ListSwarms(ctx, core.SwarmStatusActive)
if err != nil {
return fmt.Sprintf("❌ Failed to list swarms: %v", err)
}
out := "Active Swarms:\n" out := "Active Swarms:\n"
for _, sw := range swarms { out += fmt.Sprintf("- %s: %s\n", sw.ID, sw.Goal) } for _, sw := range swarms { out += fmt.Sprintf("- %s: %s\n", sw.ID, sw.Goal) }
return out return out
@ -75,3 +83,9 @@ func (s *Service) HandleCommand(ctx context.Context, input string) string {
} }
return "Unknown command" return "Unknown command"
} }
func (s *Service) Stop() {
if s.Orchestrator != nil {
s.Orchestrator.StopAll()
}
}