agents_and_robots/cmd/launcher/main.go

// Command launcher starts one or more agents from their config files.
//
// Usage:
//
//	go run ./cmd/launcher                              # auto-discovers agents/*/config.yaml
//	go run ./cmd/launcher -c agents/assistant/config.yaml
package main

import (
	"context"
	"log/slog"
	"os"
	"os/signal"
	"path/filepath"
	"sync"
	"syscall"
	"time"

	"maunium.net/go/mautrix"

	"github.com/spf13/cobra"

	"github.com/enmanuel/agents/agents"
	assistantagent "github.com/enmanuel/agents/agents/assistant-bot"
	asistente2agent "github.com/enmanuel/agents/agents/asistente-2"
	meteorologoagent "github.com/enmanuel/agents/agents/meteorologo"
	"github.com/enmanuel/agents/internal/config"
	"github.com/enmanuel/agents/pkg/decision"
	"github.com/enmanuel/agents/pkg/orchestration"
	"github.com/enmanuel/agents/shell/bus"
	agentlog "github.com/enmanuel/agents/shell/logger"
	orchshell "github.com/enmanuel/agents/shell/orchestration"
)

// rulesRegistry maps agent IDs to their rule factories.
// Add a new entry here when you create a new agent package.
var rulesRegistry = map[string]func() []decision.Rule{
	"assistant-bot": assistantagent.Rules,
	"asistente-2":   asistente2agent.Rules,
	"meteorologo":   meteorologoagent.Rules,
}

func main() {
	var (
		configPaths []string
		logLevel    string
		logDir      string
	)

	root := &cobra.Command{
		Use:   "launcher",
		Short: "Start Matrix agents from config files",
		PersistentPreRunE: func(cmd *cobra.Command, args []string) error {
			if len(configPaths) == 0 {
				matches, _ := filepath.Glob("agents/*/config.yaml")
				configPaths = matches
			}
			return nil
		},
		RunE: func(cmd *cobra.Command, args []string) error {
			lvl := parseLogLevel(logLevel)

			// ── Launcher-level logger ──
			logger, launcherCleanup, err := agentlog.NewAgentLogger(agentlog.LoggerConfig{
				BaseDir: logDir,
				AgentID: "launcher",
				Level:   lvl,
			})
			if err != nil {
				// Fallback to stdout if file logger fails.
				logger = newLogger(logLevel)
				logger.Warn("could not create file logger, falling back to stdout", "err", err)
				launcherCleanup = func() {}
			}
			var cleanups []func()
			cleanups = append(cleanups, launcherCleanup)
			defer func() {
				for _, fn := range cleanups {
					fn()
				}
			}()

			if len(configPaths) == 0 {
				logger.Warn("no agent configs found — nothing to start")
				return nil
			}

			ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
			defer stop()

			// ── Shared bus for inter-agent communication ──
			agentBus := bus.New(logger)

			// ── Start special agents (orchestrator, etc.) BEFORE normal bots ──
			orch, err := startOrchestrator(agentBus, logger)
			if err != nil {
				// Non-fatal: orchestration is optional
				logger.Warn("orchestrator not started", "err", err)
			} else {
				logger.Info("orchestrator initialized")
			}

			// ── Start normal agents ──
			var wg sync.WaitGroup
			var scannerOnce sync.Once
			var scanner *mautrix.Client

			for _, path := range configPaths {
				path := path
				cfg, err := config.Load(path)
				if err != nil {
					logger.Error("failed to load config", "path", path, "err", err)
					continue
				}
				if !cfg.Agent.Enabled {
					logger.Info("agent disabled, skipping", "id", cfg.Agent.ID)
					continue
				}

				rules := rulesFor(cfg.Agent.ID, logger)

				// Per-agent logger → writes to logs/<agent-id>/YYYY-MM-DD.jsonl
				agentLogger, agentCleanup, aErr := agentlog.NewAgentLogger(agentlog.LoggerConfig{
					BaseDir: logDir,
					AgentID: cfg.Agent.ID,
					Level:   lvl,
				})
				if aErr != nil {
					logger.Warn("agent file logger failed, using launcher logger", "agent", cfg.Agent.ID, "err", aErr)
					agentLogger = logger.With("agent", cfg.Agent.ID)
					agentCleanup = func() {}
				}
				cleanups = append(cleanups, agentCleanup)

				a, err := agents.New(cfg, rules, agentLogger)
				if err != nil {
					logger.Error("failed to create agent", "id", cfg.Agent.ID, "err", err)
					continue
				}

				// Connect agent to bus for orchestration
				a.SetBus(agentBus)

				// If orchestrator is active, wire interceptor and membership notify
				if orch != nil {
					a.SetInterceptor(orch.orchestrator.Intercept)
					a.SetMembershipNotify(orch.orchestrator.NotifyMembership)

					orch.orchestrator.RegisterParticipant(orchestration.ParticipantInfo{
						ID:           cfg.Agent.ID,
						MatrixUserID: cfg.Matrix.UserID,
						Description:  cfg.Agent.Description,
						Capabilities: cfg.Agent.Tags,
					})

					// Grab the first available Matrix client for room scanning
					scannerOnce.Do(func() {
						scanner = a.RawMatrixClient()
					})
				}

				wg.Add(1)
				go func() {
					defer wg.Done()
					agentLogger.Info("agent running")
					if err := a.Run(ctx); err != nil {
						agentLogger.Error("agent stopped with error", "err", err)
					}
				}()
			}

			// ── Startup room scan (after all participants are registered) ──
			if orch != nil && scanner != nil {
				orch.orchestrator.SetScanner(scanner)
				scanCtx, scanCancel := context.WithTimeout(ctx, 30*time.Second)
				orch.orchestrator.ScanExistingRooms(scanCtx)
				scanCancel()
			}

			wg.Wait()
			logger.Info("all agents stopped")
			return nil
		},
	}

	root.Flags().StringSliceVarP(&configPaths, "config", "c", nil,
		"Agent config file(s). If omitted, discovers all agents/*/config.yaml")
	root.Flags().StringVar(&logLevel, "log-level", "info",
		"Log level: debug | info | warn | error")
	root.Flags().StringVar(&logDir, "log-dir", "logs",
		`Log directory (logs/<agent>/YYYY-MM-DD.jsonl). Use "stdout" for console only`)

	if err := root.Execute(); err != nil {
		os.Exit(1)
	}
}

// orchHandle wraps a running orchestrator with its config for the launcher.
type orchHandle struct {
	orchestrator *orchshell.Orchestrator
	cfg          *config.SpecialConfig
}

// startOrchestrator scans agents/specials/orchestrator/config.yaml and
// initializes the orchestrator if found and enabled.
func startOrchestrator(agentBus *bus.Bus, logger *slog.Logger) (*orchHandle, error) {
	cfgPath := filepath.Join("agents", "specials", "orchestrator", "config.yaml")
	if _, err := os.Stat(cfgPath); os.IsNotExist(err) {
		return nil, err
	}

	cfg, err := config.LoadSpecial(cfgPath)
	if err != nil {
		return nil, err
	}
	if !cfg.Special.Enabled {
		return nil, nil
	}

	orchLogger := logger.With("component", "orchestrator")
	orch, err := orchshell.New(cfg, agentBus, orchLogger)
	if err != nil {
		return nil, err
	}

	return &orchHandle{orchestrator: orch, cfg: cfg}, nil
}

func rulesFor(agentID string, logger *slog.Logger) []decision.Rule {
	factory, ok := rulesRegistry[agentID]
	if !ok {
		logger.Warn("no rules registered for agent, using empty ruleset", "id", agentID)
		return nil
	}
	return factory()
}

func parseLogLevel(level string) slog.Level {
	switch level {
	case "debug":
		return slog.LevelDebug
	case "warn":
		return slog.LevelWarn
	case "error":
		return slog.LevelError
	default:
		return slog.LevelInfo
	}
}

// newLogger creates a stdout-only JSON logger (fallback when file logger fails).
func newLogger(level string) *slog.Logger {
	return slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: parseLogLevel(level)}))
}