yao/agent/robot/manager/manager.go

package manager

import (
	"context"
	"fmt"
	"sync"
	"time"

	"github.com/yaoapp/yao/agent/robot/cache"
	"github.com/yaoapp/yao/agent/robot/events"
	"github.com/yaoapp/yao/agent/robot/executor"
	"github.com/yaoapp/yao/agent/robot/pool"
	"github.com/yaoapp/yao/agent/robot/trigger"
	"github.com/yaoapp/yao/agent/robot/types"
	"github.com/yaoapp/yao/event"
	oauthtypes "github.com/yaoapp/yao/openapi/oauth/types"
)

// Default configuration values
const (
	DefaultTickInterval = time.Minute // default tick interval for clock checking
)

// Config holds manager configuration
type Config struct {
	TickInterval time.Duration  // how often to check clock triggers (default: 1 minute)
	PoolConfig   *pool.Config   // worker pool configuration
	Executor     types.Executor // optional: custom executor (default: real executor)
}

// DefaultConfig returns default manager configuration
func DefaultConfig() *Config {
	return &Config{
		TickInterval: DefaultTickInterval,
		PoolConfig:   pool.DefaultConfig(),
	}
}

// Manager implements types.Manager interface
// Orchestrates the robot scheduling system: Cache -> Dedup -> Pool -> Executor
type Manager struct {
	config   *Config
	cache    *cache.Cache
	pool     *pool.Pool
	executor types.Executor

	// Execution control for pause/resume/stop
	execController *trigger.ExecutionController

	// Ticker for clock trigger checking
	ticker     *time.Ticker
	tickerDone chan struct{}

	// State
	started bool
	mu      sync.RWMutex

	// Context for background operations
	ctx    context.Context
	cancel context.CancelFunc
}

// New creates a new manager instance with default configuration
func New() *Manager {
	return NewWithConfig(nil)
}

// NewWithConfig creates a new manager instance with custom configuration
func NewWithConfig(config *Config) *Manager {
	if config == nil {
		config = DefaultConfig()
	}

	// Apply defaults for zero values
	if config.TickInterval <= 0 {
		config.TickInterval = DefaultTickInterval
	}

	// Create components
	c := cache.New()
	p := pool.NewWithConfig(config.PoolConfig)
	ec := trigger.NewExecutionController()

	// Use custom executor if provided, otherwise create default
	var e types.Executor
	if config.Executor != nil {
		e = config.Executor
	} else {
		e = executor.New()
	}

	// Wire up pool with executor
	p.SetExecutor(e)

	// Create shared executor instances for each mode
	// These are reused across all executions to maintain accurate counters
	dryRunExecutor := executor.NewDryRun()

	// Set executor factory for mode-specific executors
	p.SetExecutorFactory(func(mode types.ExecutorMode) types.Executor {
		switch mode {
		case types.ExecutorDryRun:
			return dryRunExecutor
		case types.ExecutorSandbox:
			// Sandbox not implemented, fall back to DryRun
			return dryRunExecutor
		default:
			// Standard mode or empty - use the configured executor
			return e
		}
	})

	return &Manager{
		config:         config,
		cache:          c,
		pool:           p,
		executor:       e,
		execController: ec,
	}
}

// Start starts the manager
// 1. Load robots into cache
// 2. Start worker pool
// 3. Start clock ticker goroutine
func (m *Manager) Start() error {
	m.mu.Lock()
	defer m.mu.Unlock()

	if m.started {
		return fmt.Errorf("manager already started")
	}

	// Create background context
	m.ctx, m.cancel = context.WithCancel(context.Background())

	// Load robots into cache
	ctx := types.NewContext(m.ctx, nil)
	if err := m.cache.Load(ctx); err != nil {
		return fmt.Errorf("failed to load robots: %w", err)
	}

	// Recover non-terminal executions from previous server lifecycle
	pendingNotifications := m.recoverExecutions(m.ctx)

	// Set completion callback to clean up ExecutionController when execution finishes
	m.pool.SetOnComplete(func(execID, memberID string, status types.ExecStatus) {
		// Remove from ExecutionController (cleans up in-memory tracking)
		m.execController.Untrack(execID)
		// Remove from robot's in-memory execution list
		if robot := m.cache.Get(memberID); robot != nil {
			robot.RemoveExecution(execID)
		}
	})

	// Start worker pool
	if err := m.pool.Start(); err != nil {
		return fmt.Errorf("failed to start pool: %w", err)
	}

	// Start clock ticker
	m.ticker = time.NewTicker(m.config.TickInterval)
	m.tickerDone = make(chan struct{})

	go m.tickerLoop()

	// Start cache auto-refresh (every hour)
	m.cache.StartAutoRefresh(ctx, nil)

	m.started = true

	if len(pendingNotifications) > 0 {
		go func() {
			for _, n := range pendingNotifications {
				_, _ = event.Push(context.Background(), events.ExecRecovered, n)
			}
		}()
	}

	return nil
}

// Stop stops the manager gracefully
// 1. Stop clock ticker
// 2. Stop cache auto-refresh
// 3. Stop worker pool (waits for running jobs)
func (m *Manager) Stop() error {
	m.mu.Lock()
	if !m.started {
		m.mu.Unlock()
		return nil
	}
	m.started = false
	m.mu.Unlock()

	// Stop ticker
	if m.tickerDone != nil {
		close(m.tickerDone)
	}

	// Stop cache auto-refresh
	m.cache.StopAutoRefresh()

	// Stop pool (waits for running jobs)
	if err := m.pool.Stop(); err != nil {
		return fmt.Errorf("failed to stop pool: %w", err)
	}

	// Cancel background context
	if m.cancel != nil {
		m.cancel()
	}

	return nil
}

// tickerLoop is the main ticker goroutine
func (m *Manager) tickerLoop() {
	for {
		select {
		case <-m.tickerDone:
			m.ticker.Stop()
			return
		case now := <-m.ticker.C:
			// Perform tick - context is created per-robot in Tick()
			_ = m.Tick(m.ctx, now)
		}
	}
}

// Tick processes a clock tick
// 1. Get all cached robots
// 2. For each robot with clock trigger enabled
// 3. Check if should execute based on clock config
// 4. Submit to pool with robot's own identity
func (m *Manager) Tick(parentCtx context.Context, now time.Time) error {
	m.mu.RLock()
	if !m.started {
		m.mu.RUnlock()
		return nil
	}
	m.mu.RUnlock()

	// Get autonomous robots for clock trigger check
	robots := m.cache.ListAutonomous()

	for _, robot := range robots {
		// Skip if robot is not active
		if robot.Status == types.RobotPaused || robot.Status == types.RobotError || robot.Status == types.RobotMaintenance {
			continue
		}

		// Skip if clock trigger is disabled
		if robot.Config == nil || robot.Config.Triggers == nil {
			continue
		}
		if !robot.Config.Triggers.IsEnabled(types.TriggerClock) {
			continue
		}

		// Skip if no clock config
		if robot.Config.Clock == nil {
			continue
		}

		// Check if should trigger based on clock config
		if !m.shouldTrigger(robot, now) {
			continue
		}

		// TODO: dedup check (Phase 11.1)
		// result, err := m.dedup.Check(ctx, robot.MemberID, types.TriggerClock)
		// if err != nil || result == types.DedupSkip {
		//     continue
		// }

		// Pre-generate execution ID
		execID := pool.GenerateExecID()

		// Pre-acquire execution slot to prevent daemon-mode race condition:
		// Without this, CanRun() stays true between Tick and worker dequeue,
		// causing duplicate submissions on every tick interval.
		preExec := &types.Execution{
			ID:          execID,
			MemberID:    robot.MemberID,
			TeamID:      robot.TeamID,
			TriggerType: types.TriggerClock,
			Status:      types.ExecPending,
			StartTime:   now,
		}
		if !robot.TryAcquireSlot(preExec) {
			continue
		}

		// Track for pause/resume/stop — after slot is acquired
		ctrlExec := m.execController.Track(execID, robot.MemberID, robot.TeamID)

		// Create context with robot's own identity and cancellable context
		robotAuth := m.buildRobotAuth(robot)
		execCtx := types.NewContext(ctrlExec.Context(), robotAuth)

		// Create clock context for P0 inspiration
		clockCtx := types.NewClockContext(now, robot.Config.Clock.TZ)

		// Submit to pool with the cancellable context and execution control
		_, err := m.pool.SubmitWithID(execCtx, robot, types.TriggerClock, clockCtx, execID, ctrlExec)
		if err != nil {
			robot.RemoveExecution(execID)
			m.execController.Untrack(execID)
			continue
		}

		// Update robot's last run time
		robot.LastRun = now
	}

	return nil
}

// buildRobotAuth creates AuthorizedInfo for a robot's own identity
// Used when robot executes autonomously (clock trigger)
func (m *Manager) buildRobotAuth(robot *types.Robot) *oauthtypes.AuthorizedInfo {
	return &oauthtypes.AuthorizedInfo{
		UserID: robot.MemberID,
		TeamID: robot.TeamID,
		// ClientID could be set to a special "robot-agent" identifier if needed
		ClientID: "robot-agent",
	}
}

// shouldTrigger checks if a robot should be triggered based on its clock config
func (m *Manager) shouldTrigger(robot *types.Robot, now time.Time) bool {
	clock := robot.Config.Clock
	if clock == nil {
		return false
	}

	// Get time in robot's timezone
	loc := clock.GetLocation()
	localNow := now.In(loc)

	switch clock.Mode {
	case types.ClockTimes:
		return m.shouldTriggerTimes(robot, clock, localNow)
	case types.ClockInterval:
		return m.shouldTriggerInterval(robot, clock, localNow)
	case types.ClockDaemon:
		return m.shouldTriggerDaemon(robot, clock, localNow)
	default:
		return false
	}
}

// shouldTriggerTimes checks if current time matches any configured times
// times mode: run at specific times (e.g., ["09:00", "14:00", "17:00"])
func (m *Manager) shouldTriggerTimes(robot *types.Robot, clock *types.Clock, now time.Time) bool {
	// Check day of week first
	if !m.matchesDay(clock, now) {
		return false
	}

	// Check if current time matches any configured time
	currentTime := now.Format("15:04")
	for _, t := range clock.Times {
		if t == currentTime {
			// Check if already triggered in this minute
			if !robot.LastRun.IsZero() {
				lastRunInLoc := robot.LastRun.In(now.Location())
				if lastRunInLoc.Format("15:04") == currentTime && lastRunInLoc.Day() == now.Day() {
					return false // Already triggered this minute today
				}
			}
			return true
		}
	}
	return false
}

// shouldTriggerInterval checks if enough time has passed since last run
// interval mode: run every X duration (e.g., "30m", "2h")
func (m *Manager) shouldTriggerInterval(robot *types.Robot, clock *types.Clock, now time.Time) bool {
	interval, err := time.ParseDuration(clock.Every)
	if err != nil {
		return false
	}

	// First run if never executed
	if robot.LastRun.IsZero() {
		return true
	}

	// Check if interval has passed
	return now.Sub(robot.LastRun) >= interval
}

// shouldTriggerDaemon checks if robot can restart immediately after last run
// daemon mode: restart immediately after each run completes
func (m *Manager) shouldTriggerDaemon(robot *types.Robot, clock *types.Clock, now time.Time) bool {
	// Daemon mode: trigger if not currently running
	// CanRun() checks if robot has available execution slots
	return robot.CanRun()
}

// matchesDay checks if current day matches the configured days
func (m *Manager) matchesDay(clock *types.Clock, now time.Time) bool {
	// Empty days or ["*"] means all days
	if len(clock.Days) == 0 {
		return true
	}

	for _, day := range clock.Days {
		if day == "*" {
			return true
		}
		// Match day name (Mon, Tue, Wed, Thu, Fri, Sat, Sun)
		// or full name (Monday, Tuesday, etc.)
		weekday := now.Weekday().String()
		shortDay := weekday[:3] // Mon, Tue, etc.
		if day == weekday || day == shortDay {
			return true
		}
	}
	return false
}

// TriggerManual manually triggers a robot execution (for testing or API calls)
// This bypasses clock checking and directly submits to pool
// For non-autonomous robots: lazy-loads from DB, executes, then unloads
func (m *Manager) TriggerManual(ctx *types.Context, memberID string, trigger types.TriggerType, data interface{}) (string, error) {
	m.mu.RLock()
	if !m.started {
		m.mu.RUnlock()
		return "", fmt.Errorf("manager not started")
	}
	m.mu.RUnlock()

	// Get robot from cache, or lazy-load if not found
	robot, lazyLoaded, err := m.getOrLoadRobot(ctx, memberID)
	if err != nil {
		return "", err
	}

	// Check robot status
	if robot.Status == types.RobotPaused {
		return "", types.ErrRobotPaused
	}

	// Check if trigger type is enabled
	if robot.Config != nil && robot.Config.Triggers != nil {
		if !robot.Config.Triggers.IsEnabled(trigger) {
			return "", types.ErrTriggerDisabled
		}
	}

	// Pre-generate execution ID and track for pause/resume/stop
	// We need to track BEFORE submit so we can pass the cancellable context to the executor
	execID := pool.GenerateExecID()
	ctrlExec := m.execController.Track(execID, memberID, robot.TeamID)

	// Create a new context with the cancellable context from ExecutionController
	// This allows Stop() to propagate cancellation to the executor
	execCtx := types.NewContext(ctrlExec.Context(), ctx.Auth)
	execCtx.Locale = ctx.Locale

	// Submit to pool with the cancellable context and execution control
	// The control interface allows executor to check pause state and wait if paused
	_, err = m.pool.SubmitWithID(execCtx, robot, trigger, data, execID, ctrlExec)
	if err != nil {
		// If submission failed, untrack the execution
		m.execController.Untrack(execID)
		// If lazy-loaded and submission failed, remove from cache
		if lazyLoaded {
			m.cache.Remove(memberID)
		}
		return "", err
	}

	// For lazy-loaded robots, schedule cleanup after execution completes
	if lazyLoaded {
		m.scheduleCleanup(robot)
	}

	return execID, nil
}

// ==================== Human Intervention & Event Triggers ====================

// Intervene processes a human intervention request
// Human intervention skips P0 (inspiration) and goes directly to P1 (goals)
// For non-autonomous robots: lazy-loads from DB, executes, then unloads
func (m *Manager) Intervene(ctx *types.Context, req *types.InterveneRequest) (*types.ExecutionResult, error) {
	m.mu.RLock()
	if !m.started {
		m.mu.RUnlock()
		return nil, fmt.Errorf("manager not started")
	}
	m.mu.RUnlock()

	// Validate request
	if err := trigger.ValidateIntervention(req); err != nil {
		return nil, err
	}

	// Get robot from cache, or lazy-load if not found
	robot, lazyLoaded, err := m.getOrLoadRobot(ctx, req.MemberID)
	if err != nil {
		return nil, err
	}

	// Check robot status
	if robot.Status == types.RobotPaused {
		return nil, types.ErrRobotPaused
	}

	// Check if human trigger is enabled
	if robot.Config != nil && robot.Config.Triggers != nil {
		if !robot.Config.Triggers.IsEnabled(types.TriggerHuman) {
			return nil, types.ErrTriggerDisabled
		}
	}

	// Build trigger input
	triggerInput := &types.TriggerInput{
		Action:   req.Action,
		Messages: req.Messages,
		UserID:   ctx.UserID(),
		Locale:   req.Locale,
	}

	// Handle plan.add action - schedule for later
	if req.Action == types.ActionPlanAdd && req.PlanTime != nil {
		// If lazy-loaded but not executing, remove immediately
		if lazyLoaded {
			m.cache.Remove(req.MemberID)
		}
		// TODO: Add to plan queue (Phase 11.3)
		return &types.ExecutionResult{
			Status:  types.ExecPending,
			Message: fmt.Sprintf("Planned for %s (plan queue not implemented yet)", req.PlanTime.Format(time.RFC3339)),
		}, nil
	}

	// Determine executor mode: request > robot config > default
	executorMode := m.resolveExecutorMode(req.ExecutorMode, robot)

	// Submit to pool with executor mode
	execID, err := m.pool.SubmitWithMode(ctx, robot, types.TriggerHuman, triggerInput, executorMode)
	if err != nil {
		// If lazy-loaded and submission failed, remove from cache
		if lazyLoaded {
			m.cache.Remove(req.MemberID)
		}
		return nil, err
	}

	// Track execution for pause/resume/stop
	m.execController.Track(execID, req.MemberID, req.TeamID)

	// For lazy-loaded robots, schedule cleanup after execution completes
	if lazyLoaded {
		m.scheduleCleanup(robot)
	}

	return &types.ExecutionResult{
		ExecutionID: execID,
		Status:      types.ExecPending,
		Message:     fmt.Sprintf("Human intervention (%s) submitted", req.Action),
	}, nil
}

// HandleEvent processes an event trigger request
// Event trigger skips P0 (inspiration) and goes directly to P1 (goals)
// For non-autonomous robots: lazy-loads from DB, executes, then unloads
func (m *Manager) HandleEvent(ctx *types.Context, req *types.EventRequest) (*types.ExecutionResult, error) {
	m.mu.RLock()
	if !m.started {
		m.mu.RUnlock()
		return nil, fmt.Errorf("manager not started")
	}
	m.mu.RUnlock()

	// Validate request
	if err := trigger.ValidateEvent(req); err != nil {
		return nil, err
	}

	// Get robot from cache, or lazy-load if not found
	robot, lazyLoaded, err := m.getOrLoadRobot(ctx, req.MemberID)
	if err != nil {
		return nil, err
	}

	// Check robot status
	if robot.Status == types.RobotPaused {
		return nil, types.ErrRobotPaused
	}

	// Check if event trigger is enabled
	if robot.Config != nil && robot.Config.Triggers != nil {
		if !robot.Config.Triggers.IsEnabled(types.TriggerEvent) {
			return nil, types.ErrTriggerDisabled
		}
	}

	// Build trigger input
	triggerInput := trigger.BuildEventInput(req)

	// Determine executor mode: request > robot config > default
	executorMode := m.resolveExecutorMode(req.ExecutorMode, robot)

	// Submit to pool with executor mode
	execID, err := m.pool.SubmitWithMode(ctx, robot, types.TriggerEvent, triggerInput, executorMode)
	if err != nil {
		// If lazy-loaded and submission failed, remove from cache
		if lazyLoaded {
			m.cache.Remove(req.MemberID)
		}
		return nil, err
	}

	// Track execution for pause/resume/stop
	m.execController.Track(execID, req.MemberID, "")

	// For lazy-loaded robots, schedule cleanup after execution completes
	if lazyLoaded {
		m.scheduleCleanup(robot)
	}

	return &types.ExecutionResult{
		ExecutionID: execID,
		Status:      types.ExecPending,
		Message:     fmt.Sprintf("Event trigger (%s: %s) submitted", req.Source, req.EventType),
	}, nil
}

// ==================== Execution Control ====================

// PauseExecution pauses a running execution
func (m *Manager) PauseExecution(ctx *types.Context, execID string) error {
	// Get execution info before pausing
	exec := m.execController.Get(execID)
	if exec == nil {
		return fmt.Errorf("execution not found: %s", execID)
	}

	// Pause the execution
	if err := m.execController.Pause(execID); err != nil {
		return err
	}

	// Remove from robot's in-memory execution list (paused doesn't count as running)
	if robot := m.cache.Get(exec.MemberID); robot != nil {
		robot.RemoveExecution(execID)
	}

	return nil
}

// ResumeExecution resumes a paused execution
func (m *Manager) ResumeExecution(ctx *types.Context, execID string) error {
	// Get execution info before resuming
	exec := m.execController.Get(execID)
	if exec == nil {
		return fmt.Errorf("execution not found: %s", execID)
	}

	// Resume the execution
	if err := m.execController.Resume(execID); err != nil {
		return err
	}

	// Add back to robot's in-memory execution list
	if robot := m.cache.Get(exec.MemberID); robot != nil {
		robot.AddExecution(&types.Execution{
			ID:       execID,
			MemberID: exec.MemberID,
			TeamID:   exec.TeamID,
			Status:   types.ExecRunning,
		})
	}

	return nil
}

// StopExecution stops a running execution
func (m *Manager) StopExecution(ctx *types.Context, execID string) error {
	// Get execution info before stopping
	exec := m.execController.Get(execID)
	if exec == nil {
		return fmt.Errorf("execution not found: %s", execID)
	}

	// Stop the execution
	if err := m.execController.Stop(execID); err != nil {
		return err
	}

	// Remove from robot's in-memory execution list
	if robot := m.cache.Get(exec.MemberID); robot != nil {
		robot.RemoveExecution(execID)
	}

	return nil
}

// GetExecutionStatus returns the status of an execution
func (m *Manager) GetExecutionStatus(execID string) (*trigger.ControlledExecution, error) {
	exec := m.execController.Get(execID)
	if exec == nil {
		return nil, fmt.Errorf("execution not found: %s", execID)
	}
	return exec, nil
}

// ListExecutions returns all tracked executions
func (m *Manager) ListExecutions() []*trigger.ControlledExecution {
	return m.execController.List()
}

// ListExecutionsByMember returns all executions for a specific robot
func (m *Manager) ListExecutionsByMember(memberID string) []*trigger.ControlledExecution {
	return m.execController.ListByMember(memberID)
}

// ==================== Helper Methods ====================

// getOrLoadRobot gets a robot from cache, or lazy-loads from DB if not found
// Returns: robot, wasLazyLoaded, error
func (m *Manager) getOrLoadRobot(ctx *types.Context, memberID string) (*types.Robot, bool, error) {
	// Try cache first
	robot := m.cache.Get(memberID)
	if robot != nil {
		return robot, false, nil
	}

	// Not in cache - lazy load from database
	robot, err := m.cache.LoadByID(ctx, memberID)
	if err != nil {
		return nil, false, err
	}

	// Add to cache temporarily for execution tracking
	m.cache.Add(robot)

	// Return with lazyLoaded=true to indicate cleanup needed after execution
	return robot, true, nil
}

// scheduleCleanup schedules removal of a lazy-loaded robot after all executions complete
// This runs in a goroutine that monitors the robot's execution count
func (m *Manager) scheduleCleanup(robot *types.Robot) {
	go func() {
		memberID := robot.MemberID

		// Poll every 5 seconds to check if all executions are done
		ticker := time.NewTicker(5 * time.Second)
		defer ticker.Stop()

		// Timeout after 24 hours to prevent memory leaks
		timeout := time.After(24 * time.Hour)

		for {
			select {
			case <-timeout:
				// Timeout - force cleanup
				m.cache.Remove(memberID)
				return

			case <-ticker.C:
				// Check if robot still exists in cache
				r := m.cache.Get(memberID)
				if r == nil {
					// Already removed
					return
				}

				// Check if all executions are done
				if r.RunningCount() == 0 {
					// Non-autonomous robots: with full-cache load they will be
					// re-added on next Load() cycle, so removal is a no-op in practice.
					if !r.AutonomousMode {
						m.cache.Remove(memberID)
					}
					return
				}
			}
		}
	}()
}

// resolveExecutorMode determines the executor mode to use
// Priority: request > robot config > default (standard)
func (m *Manager) resolveExecutorMode(requestMode types.ExecutorMode, robot *types.Robot) types.ExecutorMode {
	// Request mode takes precedence
	if requestMode != "" && requestMode.IsValid() {
		return requestMode
	}

	// Robot config mode
	if robot != nil && robot.Config != nil && robot.Config.Executor != nil {
		return robot.Config.Executor.GetMode()
	}

	// Default: standard
	return types.ExecutorStandard
}

// ==================== Getters for internal components ====================
// These are exposed for testing and advanced use cases

// Cache returns the internal cache
func (m *Manager) Cache() *cache.Cache {
	return m.cache
}

// Pool returns the internal pool
func (m *Manager) Pool() *pool.Pool {
	return m.pool
}

// Executor returns the internal executor
func (m *Manager) Executor() types.Executor {
	return m.executor
}

// ExecController returns the internal execution controller
func (m *Manager) ExecController() *trigger.ExecutionController {
	return m.execController
}

// IsStarted returns true if manager is started
func (m *Manager) IsStarted() bool {
	m.mu.RLock()
	defer m.mu.RUnlock()
	return m.started
}

// Running returns number of currently running jobs
func (m *Manager) Running() int {
	return m.pool.Running()
}

// Queued returns number of queued jobs
func (m *Manager) Queued() int {
	return m.pool.Queued()
}

// CachedRobots returns number of cached robots
func (m *Manager) CachedRobots() int {
	return m.cache.Count()
}