diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index b3fa4faf0..f7656110f 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -133,6 +133,9 @@ func (al *AgentLoop) Run(ctx context.Context) error { func (al *AgentLoop) Stop() { al.running = false + if al.swarm != nil { + al.swarm.Stop() + } } func (al *AgentLoop) ProcessDirect(ctx context.Context, content, sessionKey string) (string, error) { diff --git a/pkg/swarm/config/config.go b/pkg/swarm/config/config.go index ca2821873..5347028e4 100644 --- a/pkg/swarm/config/config.go +++ b/pkg/swarm/config/config.go @@ -15,12 +15,13 @@ type MemoryConfig struct { } type LimitsConfig struct { - MaxDepth int `json:"max_depth" env:"PICOCLAW_SWARM_MAX_DEPTH"` - MaxNodes int `json:"max_nodes" env:"PICOCLAW_SWARM_MAX_NODES"` - GlobalTimeout time.Duration `json:"global_timeout" env:"PICOCLAW_SWARM_GLOBAL_TIMEOUT"` - MaxRetries int `json:"max_retries" env:"PICOCLAW_SWARM_MAX_RETRIES"` - MaxIterations int `json:"max_iterations" env:"PICOCLAW_SWARM_MAX_ITERATIONS"` - PruningMsgKeep int `json:"pruning_msg_keep" env:"PICOCLAW_SWARM_PRUNING_MSG_KEEP"` + MaxDepth int `json:"max_depth" env:"PICOCLAW_SWARM_MAX_DEPTH"` + MaxNodes int `json:"max_nodes" env:"PICOCLAW_SWARM_MAX_NODES"` + GlobalTimeout time.Duration `json:"global_timeout" env:"PICOCLAW_SWARM_GLOBAL_TIMEOUT"` + MaxRetries int `json:"max_retries" env:"PICOCLAW_SWARM_MAX_RETRIES"` + MaxIterations int `json:"max_iterations" env:"PICOCLAW_SWARM_MAX_ITERATIONS"` + 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 { @@ -44,12 +45,13 @@ type Policy struct { func DefaultSwarmConfig() SwarmConfig { return SwarmConfig{ Limits: LimitsConfig{ - MaxDepth: 3, - MaxNodes: 10, - GlobalTimeout: 10 * time.Minute, - MaxRetries: 3, - MaxIterations: 10, - PruningMsgKeep: 6, // Keep last 6 messages when pruning + MaxDepth: 3, + MaxNodes: 10, + GlobalTimeout: 10 * time.Minute, + MaxRetries: 3, + MaxIterations: 10, + PruningMsgKeep: 6, // Keep last 6 messages when pruning + SummarizeThreshold: 20, // Summarize after 20 messages }, Resilience: ResilienceConfig{ RetryBackoff: 2 * time.Second, diff --git a/pkg/swarm/memory/chromem_store.go b/pkg/swarm/memory/chromem_store.go index e0609dcde..f9bceb917 100644 --- a/pkg/swarm/memory/chromem_store.go +++ b/pkg/swarm/memory/chromem_store.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "runtime" + "time" "github.com/philippgille/chromem-go" "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 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, Metadata: meta, } diff --git a/pkg/swarm/memory/sqlite_store.go b/pkg/swarm/memory/sqlite_store.go index c50c059ee..57d9a8800 100644 --- a/pkg/swarm/memory/sqlite_store.go +++ b/pkg/swarm/memory/sqlite_store.go @@ -17,14 +17,23 @@ func NewSQLiteStore(dbPath string) (*SQLiteStore, error) { db, err := sql.Open("sqlite", dbPath) if err != nil { return nil, err } s := &SQLiteStore{db: db} - s.init() + if err := s.init(); err != nil { + return nil, err + } return s, nil } -func (s *SQLiteStore) init() { - s.db.Exec(`CREATE TABLE IF NOT EXISTS swarms (id TEXT PRIMARY KEY, goal TEXT, status TEXT, created_at DATETIME);`) - 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);`) - s.db.Exec(`CREATE TABLE IF NOT EXISTS facts (swarm_id TEXT, content TEXT, confidence REAL, source TEXT, metadata JSON);`) +func (s *SQLiteStore) init() error { + if _, err := s.db.Exec(`CREATE TABLE IF NOT EXISTS swarms (id TEXT PRIMARY KEY, goal TEXT, status TEXT, created_at DATETIME);`); err != nil { + return err + } + 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 { @@ -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) { row := s.db.QueryRowContext(ctx, "SELECT id, goal, status, created_at FROM swarms WHERE id=?", id) var sw core.Swarm - err := row.Scan(&sw.ID, &sw.Goal, &sw.Status, &sw.CreatedAt) - return &sw, err + if err := row.Scan(&sw.ID, &sw.Goal, &sw.Status, &sw.CreatedAt); err != nil { + return nil, err + } + return &sw, nil } 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) { - 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() var list []*core.Swarm for rows.Next() { 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) } return list, nil } func (s *SQLiteStore) CreateNode(ctx context.Context, n *core.Node) error { - role, _ := json.Marshal(n.Role) - stats, _ := json.Marshal(n.Stats) - _, err := s.db.ExecContext(ctx, "INSERT INTO nodes VALUES (?, ?, ?, ?, ?, ?, ?, ?)", n.ID, n.SwarmID, n.ParentID, role, n.Task, n.Status, n.Output, stats) + role, err := json.Marshal(n.Role) + if err != nil { + 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 } @@ -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) var n core.Node var role, stats []byte - row.Scan(&n.ID, &n.SwarmID, &n.ParentID, &role, &n.Task, &n.Status, &n.Output, &stats) - json.Unmarshal(role, &n.Role) - json.Unmarshal(stats, &n.Stats) + if err := row.Scan(&n.ID, &n.SwarmID, &n.ParentID, &role, &n.Task, &n.Status, &n.Output, &stats); err != nil { + return nil, err + } + 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 } func (s *SQLiteStore) UpdateNode(ctx context.Context, n *core.Node) error { - stats, _ := json.Marshal(n.Stats) - _, err := s.db.ExecContext(ctx, "UPDATE nodes SET status=?, output=?, stats=? WHERE id=?", n.Status, n.Output, stats, n.ID) + stats, err := json.Marshal(n.Stats) + 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 } 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() var list []*core.Node for rows.Next() { var n core.Node var role, stats []byte - rows.Scan(&n.ID, &n.SwarmID, &n.ParentID, &role, &n.Task, &n.Status, &n.Output, &stats) - json.Unmarshal(role, &n.Role) - json.Unmarshal(stats, &n.Stats) + if err := rows.Scan(&n.ID, &n.SwarmID, &n.ParentID, &role, &n.Task, &n.Status, &n.Output, &stats); err != nil { + return nil, err + } + 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) } return list, nil } func (s *SQLiteStore) SaveFact(ctx context.Context, f core.Fact) error { - meta, _ := json.Marshal(f.Metadata) - _, err := s.db.ExecContext(ctx, "INSERT INTO facts VALUES (?, ?, ?, ?, ?)", f.SwarmID, f.Content, f.Confidence, f.Source, meta) + meta, err := json.Marshal(f.Metadata) + 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 } @@ -109,12 +152,17 @@ func (s *SQLiteStore) SearchFacts(ctx context.Context, id core.SwarmID, q string 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() var res []core.FactResult for rows.Next() { 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}) } return res, nil diff --git a/pkg/swarm/prompt/prompts.go b/pkg/swarm/prompt/prompts.go index d6f65c577..cce1e2f54 100644 --- a/pkg/swarm/prompt/prompts.go +++ b/pkg/swarm/prompt/prompts.go @@ -13,7 +13,7 @@ SWARM CONTEXT: - Your Role: %s 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. 3. OUTPUT: Provide clear, structured reasoning. 4. TOOLS: Use available tools to gather facts. Do not guess. diff --git a/pkg/swarm/runtime/node.go b/pkg/swarm/runtime/node.go index 81a6188f7..f6b96a4ed 100644 --- a/pkg/swarm/runtime/node.go +++ b/pkg/swarm/runtime/node.go @@ -6,6 +6,7 @@ import ( "fmt" "log/slog" + "github.com/sipeed/picoclaw/pkg/swarm/config" "github.com/sipeed/picoclaw/pkg/swarm/core" "github.com/sipeed/picoclaw/pkg/tools" ) @@ -17,11 +18,12 @@ type NodeActor struct { LLM core.LLMClient Tools *tools.ToolRegistry Policy *core.PolicyChecker + Config config.SwarmConfig 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{ Data: data, Bus: bus, @@ -29,13 +31,16 @@ func NewNodeActor(data *core.Node, bus core.EventBus, store core.SwarmStore, llm LLM: llm, Tools: toolRegistry, Policy: policy, + Config: cfg, peerInsights: make([]string, 0), } } func (n *NodeActor) Start(ctx context.Context) { 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 sub, _ := n.Bus.Subscribe("node.events", func(e core.Event) { @@ -69,35 +74,52 @@ func (n *NodeActor) run(ctx context.Context) { allowedTools := n.getToolsForRole() model := n.Data.Role.Model - for i := 0; i < 10; i++ { // Max 10 iterations - // Progressive Summarization: If history > 20 messages, compress the middle part - if len(messages) > 20 { + maxIterations := n.Config.Limits.MaxIterations + if maxIterations <= 0 { + 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) // 1. Identify context to summarize // Header: System (0) + User Goal (1) - // Tail: Last 6 messages - toSummarize := messages[2 : len(messages)-6] + // Tail: Last N messages + 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." - summaryMessages := append(toSummarize, core.Message{Role: "user", Content: summaryPrompt}) + if len(messages) > keepCount+2 { + toSummarize := messages[2 : len(messages)-keepCount] - summaryResp, err := n.LLM.Chat(ctx, summaryMessages, nil, model) - if err == nil { - // 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...) + summaryPrompt := "Briefly summarize the key findings, data points, and progress from the conversation above. Focus on facts discovered so far. Be very concise." + summaryMessages := append(toSummarize, core.Message{Role: "user", Content: summaryPrompt}) - 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."}, - }) + summaryResp, err := n.LLM.Chat(ctx, summaryMessages, nil, model) + if err == nil { + // 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) { n.Data.Output = out 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{ Type: core.EventNodeCompleted, SwarmID: n.Data.SwarmID, NodeID: n.Data.ID, 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) { 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{ Type: core.EventNodeFailed, SwarmID: n.Data.SwarmID, NodeID: n.Data.ID, Payload: map[string]any{"error": err.Error()}, diff --git a/pkg/swarm/runtime/orchestrator.go b/pkg/swarm/runtime/orchestrator.go index dc06602f1..71652c657 100644 --- a/pkg/swarm/runtime/orchestrator.go +++ b/pkg/swarm/runtime/orchestrator.go @@ -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) { 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() - sCtx, cancel := context.WithCancel(context.Background()) + sCtx, cancel := context.WithCancel(ctx) o.activeSwarms[id] = cancel 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 } @@ -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}, 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() 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) sub, _ := o.bus.Subscribe("node.events", func(e core.Event) { 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"]) } } }) 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 { case out := <-done: return out, nil case err := <-fail: return "", 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) } } + +func (o *Orchestrator) StopAll() { + o.mu.Lock() + defer o.mu.Unlock() + for id, cancel := range o.activeSwarms { + cancel() + delete(o.activeSwarms, id) + } +} diff --git a/pkg/swarm/service.go b/pkg/swarm/service.go index 6310f1b3c..0c7589526 100644 --- a/pkg/swarm/service.go +++ b/pkg/swarm/service.go @@ -41,7 +41,7 @@ func NewService(dbPath string, provider providers.LLMProvider, registry *tools.T } 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 := "" switch e.Type { case core.EventNodeThinking: msg = fmt.Sprintf("🤖 [%s]: %s", e.NodeID[:4], e.Payload["content"]) @@ -51,7 +51,9 @@ func (s *Service) listen() { if msg != "" { 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 { @@ -61,10 +63,16 @@ func (s *Service) HandleCommand(ctx context.Context, input string) string { switch args[1] { case "spawn": 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) 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" for _, sw := range swarms { out += fmt.Sprintf("- %s: %s\n", sw.ID, sw.Goal) } return out @@ -74,4 +82,10 @@ func (s *Service) HandleCommand(ctx context.Context, input string) string { return "Stopped." } return "Unknown command" +} + +func (s *Service) Stop() { + if s.Orchestrator != nil { + s.Orchestrator.StopAll() + } } \ No newline at end of file