diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index f7656110f..cb87f9284 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -84,9 +84,9 @@ func (al *AgentLoop) Run(ctx context.Context) error { go func() { for msg := range al.swarm.Outbound { al.bus.PublishOutbound(bus.OutboundMessage{ - Channel: "cli", // Default to CLI for now, ideally dynamic - ChatID: "direct", - Content: msg, + Channel: msg.Channel, + ChatID: msg.ChatID, + Content: msg.Content, }) } }() @@ -104,7 +104,7 @@ func (al *AgentLoop) Run(ctx context.Context) error { // Intercept /swarm commands if al.swarm != nil && len(msg.Content) > 6 && msg.Content[:7] == "/swarm " { - response := al.swarm.HandleCommand(ctx, msg.Content) + response := al.swarm.HandleCommand(ctx, msg.Channel, msg.ChatID, msg.Content) al.bus.PublishOutbound(bus.OutboundMessage{ Channel: msg.Channel, ChatID: msg.ChatID, @@ -141,7 +141,7 @@ func (al *AgentLoop) Stop() { func (al *AgentLoop) ProcessDirect(ctx context.Context, content, sessionKey string) (string, error) { // Intercept /swarm commands if al.swarm != nil && len(content) > 6 && content[:7] == "/swarm " { - return al.swarm.HandleCommand(ctx, content), nil + return al.swarm.HandleCommand(ctx, "cli", "direct", content), nil } msg := bus.InboundMessage{ diff --git a/pkg/swarm/core/core.go b/pkg/swarm/core/core.go index ac0dfbc41..0cef97631 100644 --- a/pkg/swarm/core/core.go +++ b/pkg/swarm/core/core.go @@ -36,10 +36,12 @@ const ( // --- Core Structs --- type Swarm struct { - ID SwarmID `json:"id"` - Goal string `json:"goal"` - Status SwarmStatus `json:"status"` - CreatedAt time.Time `json:"created_at"` + ID SwarmID `json:"id"` + Goal string `json:"goal"` + Status SwarmStatus `json:"status"` + CreatedAt time.Time `json:"created_at"` + OriginChannel string `json:"origin_channel"` + OriginChatID string `json:"origin_chat_id"` } type Node struct { @@ -126,3 +128,10 @@ type FactResult struct { Content string `json:"content"` Score float32 `json:"score"` } + +// --- Relay Types --- +type RelayMessage struct { + Channel string + ChatID string + Content string +} diff --git a/pkg/swarm/memory/sqlite_store.go b/pkg/swarm/memory/sqlite_store.go index af3689ed1..7a11916d9 100644 --- a/pkg/swarm/memory/sqlite_store.go +++ b/pkg/swarm/memory/sqlite_store.go @@ -24,7 +24,7 @@ func NewSQLiteStore(dbPath string) (*SQLiteStore, error) { } 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 { + if _, err := s.db.Exec(`CREATE TABLE IF NOT EXISTS swarms (id TEXT PRIMARY KEY, goal TEXT, status TEXT, created_at DATETIME, origin_channel TEXT, origin_chat_id TEXT);`); 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 { @@ -37,14 +37,14 @@ func (s *SQLiteStore) init() error { } func (s *SQLiteStore) CreateSwarm(ctx context.Context, sw *core.Swarm) error { - _, err := s.db.ExecContext(ctx, "INSERT INTO swarms (id, goal, status, created_at) VALUES (?, ?, ?, ?)", sw.ID, sw.Goal, sw.Status, sw.CreatedAt) + _, err := s.db.ExecContext(ctx, "INSERT INTO swarms (id, goal, status, created_at, origin_channel, origin_chat_id) VALUES (?, ?, ?, ?, ?, ?)", sw.ID, sw.Goal, sw.Status, sw.CreatedAt, sw.OriginChannel, sw.OriginChatID) return err } 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, origin_channel, origin_chat_id FROM swarms WHERE id=?", id) var sw core.Swarm - if err := row.Scan(&sw.ID, &sw.Goal, &sw.Status, &sw.CreatedAt); err != nil { + if err := row.Scan(&sw.ID, &sw.Goal, &sw.Status, &sw.CreatedAt, &sw.OriginChannel, &sw.OriginChatID); err != nil { return nil, err } return &sw, nil @@ -56,7 +56,7 @@ 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, err := 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, origin_channel, origin_chat_id FROM swarms WHERE status=?", status) if err != nil { return nil, err } @@ -64,7 +64,7 @@ func (s *SQLiteStore) ListSwarms(ctx context.Context, status core.SwarmStatus) ( var list []*core.Swarm for rows.Next() { var sw core.Swarm - if err := rows.Scan(&sw.ID, &sw.Goal, &sw.Status, &sw.CreatedAt); err != nil { + if err := rows.Scan(&sw.ID, &sw.Goal, &sw.Status, &sw.CreatedAt, &sw.OriginChannel, &sw.OriginChatID); err != nil { return nil, err } list = append(list, &sw) diff --git a/pkg/swarm/runtime/orchestrator.go b/pkg/swarm/runtime/orchestrator.go index 71652c657..29866790f 100644 --- a/pkg/swarm/runtime/orchestrator.go +++ b/pkg/swarm/runtime/orchestrator.go @@ -39,9 +39,12 @@ func NewOrchestrator(store core.SwarmStore, bus core.EventBus, llm core.LLMClien 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, channel, chatID string) (core.SwarmID, error) { id := core.SwarmID(uuid.New().String()) - if err := o.store.CreateSwarm(ctx, &core.Swarm{ID: id, Goal: goal, Status: core.SwarmStatusActive, CreatedAt: time.Now()}); err != nil { + if err := o.store.CreateSwarm(ctx, &core.Swarm{ + ID: id, Goal: goal, Status: core.SwarmStatusActive, CreatedAt: time.Now(), + OriginChannel: channel, OriginChatID: chatID, + }); err != nil { return "", fmt.Errorf("failed to create swarm in store: %w", err) } diff --git a/pkg/swarm/service.go b/pkg/swarm/service.go index 80d75ee02..b26a072bd 100644 --- a/pkg/swarm/service.go +++ b/pkg/swarm/service.go @@ -19,7 +19,7 @@ type Service struct { Orchestrator *runtime.Orchestrator Store core.SwarmStore Bus core.EventBus - Outbound chan string + Outbound chan core.RelayMessage } func NewService(dbPath string, provider providers.LLMProvider, registry *tools.ToolRegistry, cfg config.SwarmConfig, model string) (*Service, error) { @@ -35,7 +35,7 @@ func NewService(dbPath string, provider providers.LLMProvider, registry *tools.T orch := runtime.NewOrchestrator(store, eventBus, adapter, registry, cfg, model) orch.SetSharedMemory(sharedMem) - s := &Service{Orchestrator: orch, Store: store, Bus: eventBus, Outbound: make(chan string, 100)} + s := &Service{Orchestrator: orch, Store: store, Bus: eventBus, Outbound: make(chan core.RelayMessage, 100)} s.listen() return s, nil } @@ -49,21 +49,42 @@ func (s *Service) listen() { case core.EventNodeFailed: msg = fmt.Sprintf("❌ [%s] Failed: %v", e.NodeID[:4], e.Payload["error"]) } if msg != "" { - select { case s.Outbound <- msg: default: } + // Get swarm origin + sw, err := s.Store.GetSwarm(context.Background(), e.SwarmID) + if err == nil && sw.OriginChannel != "" { + select { + case s.Outbound <- core.RelayMessage{ + Channel: sw.OriginChannel, + ChatID: sw.OriginChatID, + Content: msg, + }: + default: + } + } else { + // Fallback to CLI + select { + case s.Outbound <- core.RelayMessage{ + Channel: "cli", + ChatID: "direct", + Content: 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, channel, chatID, input string) string { args := strings.Fields(input) if len(args) < 2 { return "Usage: /swarm [goal/id]" } switch args[1] { case "spawn": goal := strings.Join(args[2:], " ") - id, err := s.Orchestrator.SpawnSwarm(ctx, goal) + id, err := s.Orchestrator.SpawnSwarm(ctx, goal, channel, chatID) if err != nil { return fmt.Sprintf("❌ Failed to spawn swarm: %v", err) }