yao/agent/robot/executor/standard/executor.go

package standard

import (
	"fmt"
	"strings"
	"sync/atomic"
	"time"

	kunlog "github.com/yaoapp/kun/log"
	agentcontext "github.com/yaoapp/yao/agent/context"
	robotevents "github.com/yaoapp/yao/agent/robot/events"
	"github.com/yaoapp/yao/agent/robot/executor/types"
	"github.com/yaoapp/yao/agent/robot/store"
	robottypes "github.com/yaoapp/yao/agent/robot/types"
	"github.com/yaoapp/yao/agent/robot/utils"
	"github.com/yaoapp/yao/event"
)

// Executor implements the standard executor with real Agent calls
// This is the production executor that:
// - Persists execution history to database
// - Calls real Agents via Assistant.Stream()
// - Logs phase transitions and errors using kun/log
type Executor struct {
	config       types.Config
	store        *store.ExecutionStore
	robotStore   *store.RobotStore
	execCount    atomic.Int32
	currentCount atomic.Int32
	onStart      func()
	onEnd        func()
}

// New creates a new standard executor
func New() *Executor {
	return &Executor{
		store:      store.NewExecutionStore(),
		robotStore: store.NewRobotStore(),
	}
}

// NewWithConfig creates a new standard executor with configuration
func NewWithConfig(config types.Config) *Executor {
	return &Executor{
		config:     config,
		store:      store.NewExecutionStore(),
		robotStore: store.NewRobotStore(),
	}
}

// Execute runs a robot through all applicable phases with real Agent calls (auto-generates ID)
func (e *Executor) Execute(ctx *robottypes.Context, robot *robottypes.Robot, trigger robottypes.TriggerType, data interface{}) (*robottypes.Execution, error) {
	return e.ExecuteWithControl(ctx, robot, trigger, data, "", nil)
}

// ExecuteWithID runs a robot through all applicable phases with a pre-generated execution ID (no control)
func (e *Executor) ExecuteWithID(ctx *robottypes.Context, robot *robottypes.Robot, trigger robottypes.TriggerType, data interface{}, execID string) (*robottypes.Execution, error) {
	return e.ExecuteWithControl(ctx, robot, trigger, data, execID, nil)
}

// ExecuteWithControl runs a robot through all applicable phases with execution control
// control: optional, allows pause/resume functionality during execution
func (e *Executor) ExecuteWithControl(ctx *robottypes.Context, robot *robottypes.Robot, trigger robottypes.TriggerType, data interface{}, execID string, control robottypes.ExecutionControl) (*robottypes.Execution, error) {
	if robot == nil {
		return nil, fmt.Errorf("robot cannot be nil")
	}

	// Determine starting phase based on trigger type
	startPhaseIndex := 0
	if trigger == robottypes.TriggerHuman || trigger == robottypes.TriggerEvent {
		startPhaseIndex = 1 // Skip P0 (Inspiration)
	}

	// Use provided execID or generate new one
	if execID == "" {
		execID = utils.NewID()
	}

	// Create execution (Job system removed, using ExecutionStore only)
	input := types.BuildTriggerInput(trigger, data)
	exec := &robottypes.Execution{
		ID:          execID,
		MemberID:    robot.MemberID,
		TeamID:      robot.TeamID,
		TriggerType: trigger,
		StartTime:   time.Now(),
		Status:      robottypes.ExecPending,
		Phase:       robottypes.AllPhases[startPhaseIndex],
		Input:       input,
		ChatID:      fmt.Sprintf("robot_%s_%s", robot.MemberID, execID),
	}

	// Load pre-existing Goals/Tasks from store when resuming a confirmed execution.
	// RunGoals and RunTasks have skip logic when these are already populated.
	if execID != "" && !e.config.SkipPersistence && e.store != nil {
		if existing, err := e.store.Get(ctx.Context, execID); err == nil && existing != nil {
			exec.Goals = existing.Goals
			exec.Tasks = existing.Tasks
			if existing.Input != nil {
				exec.Input = existing.Input
			}
		}
	}

	// If goals are pre-confirmed (passed via Input.Data["goals"]), inject them directly.
	// RunGoals will skip LLM call when exec.Goals is already populated (§18.2).
	if exec.Goals == nil && input != nil && input.Data != nil {
		if goalsStr, ok := input.Data["goals"].(string); ok && goalsStr != "" {
			exec.Goals = &robottypes.Goals{Content: goalsStr}
		}
	}

	// Initialize UI display fields (with i18n support)
	exec.Name, exec.CurrentTaskName = e.initUIFields(trigger, input, robot)

	// Set robot reference for phase methods
	exec.SetRobot(robot)

	// Persist execution record to database
	// Robot is identified by member_id (globally unique in __yao.member table)
	if !e.config.SkipPersistence && e.store != nil {
		record := store.FromExecution(exec)
		if err := e.store.Save(ctx.Context, record); err != nil {
			// Log warning but don't fail execution
			kunlog.With(kunlog.F{
				"execution_id": exec.ID,
				"member_id":    exec.MemberID,
				"error":        err,
			}).Warn("Failed to persist execution record: %v", err)
		}

		// If goals were pre-injected, persist them and update the execution title
		if exec.Goals != nil && exec.Goals.Content != "" {
			if err := e.store.UpdatePhase(ctx.Context, exec.ID, robottypes.PhaseGoals, exec.Goals); err != nil {
				kunlog.With(kunlog.F{
					"execution_id": exec.ID,
					"member_id":    exec.MemberID,
					"error":        err,
				}).Warn("Failed to persist pre-confirmed goals: %v", err)
			}
			if goalName := extractGoalName(exec.Goals); goalName != "" {
				e.updateUIFields(ctx, exec, goalName, "")
			}
		}

	}

	// Acquire execution slot
	if !robot.TryAcquireSlot(exec) {
		kunlog.With(kunlog.F{
			"execution_id": exec.ID,
			"member_id":    exec.MemberID,
		}).Warn("Execution quota exceeded")
		return nil, robottypes.ErrQuotaExceeded
	}
	// Defer: remove execution from robot's tracking (unless suspended) and update robot status
	defer func() {
		// Suspended executions stay in tracking — they are still "alive"
		if exec.Status == robottypes.ExecWaiting {
			return
		}
		robot.RemoveExecution(exec.ID)
		// Update robot status to idle if no more running executions
		if robot.RunningCount() == 0 && !e.config.SkipPersistence && e.robotStore != nil {
			if err := e.robotStore.UpdateStatus(ctx.Context, robot.MemberID, robottypes.RobotIdle); err != nil {
				kunlog.With(kunlog.F{
					"member_id": robot.MemberID,
					"error":     err,
				}).Warn("Failed to update robot status to idle: %v", err)
			}
		}
	}()

	// Track execution count
	e.execCount.Add(1)
	e.currentCount.Add(1)
	defer e.currentCount.Add(-1)

	// Callbacks
	if e.onStart != nil {
		e.onStart()
	}
	if e.onEnd != nil {
		defer e.onEnd()
	}

	// Update status to running
	exec.Status = robottypes.ExecRunning
	kunlog.With(kunlog.F{
		"execution_id": exec.ID,
		"member_id":    exec.MemberID,
		"trigger_type": string(exec.TriggerType),
	}).Info("Execution started")

	// Persist running status
	if !e.config.SkipPersistence && e.store != nil {
		if err := e.store.UpdateStatus(ctx.Context, exec.ID, robottypes.ExecRunning, ""); err != nil {
			kunlog.With(kunlog.F{
				"execution_id": exec.ID,
				"error":        err,
			}).Warn("Failed to persist running status: %v", err)
		}
	}

	// Update robot status to working (when execution starts)
	if !e.config.SkipPersistence && e.robotStore != nil {
		if err := e.robotStore.UpdateStatus(ctx.Context, robot.MemberID, robottypes.RobotWorking); err != nil {
			kunlog.With(kunlog.F{
				"member_id": robot.MemberID,
				"error":     err,
			}).Warn("Failed to update robot status to working: %v", err)
		}
	}

	// Check for simulated failure (for testing)
	if dataStr, ok := data.(string); ok && dataStr == "simulate_failure" {
		exec.Status = robottypes.ExecFailed
		exec.Error = "simulated failure"
		kunlog.With(kunlog.F{
			"execution_id": exec.ID,
			"member_id":    exec.MemberID,
		}).Warn("Simulated failure triggered")
		// Persist failed status
		if !e.config.SkipPersistence && e.store != nil {
			_ = e.store.UpdateStatus(ctx.Context, exec.ID, robottypes.ExecFailed, "simulated failure")
		}
		return exec, nil
	}

	// Determine locale for UI messages
	locale := getEffectiveLocale(robot, exec.Input)

	// Execute phases (PhaseHost is not part of the normal pipeline — it is only for Interact)
	phases := robottypes.AllPhases[startPhaseIndex:]
	for _, phase := range phases {
		if phase == robottypes.PhaseHost {
			continue
		}
		if err := e.runPhase(ctx, exec, phase, data, control); err != nil {
			// Check if execution was suspended (needs human input)
			if err == robottypes.ErrExecutionSuspended {
				kunlog.With(kunlog.F{
					"execution_id": exec.ID,
					"member_id":    exec.MemberID,
					"phase":        string(phase),
				}).Info("Execution suspended during phase %s", phase)
				return exec, robottypes.ErrExecutionSuspended
			}

			// Check if execution was cancelled
			if err == robottypes.ErrExecutionCancelled {
				exec.Status = robottypes.ExecCancelled
				exec.Error = "execution cancelled by user"
				now := time.Now()
				exec.EndTime = &now

				// Update UI field for cancellation with i18n
				e.updateUIFields(ctx, exec, "", getLocalizedMessage(locale, "cancelled"))

				kunlog.With(kunlog.F{
					"execution_id": exec.ID,
					"member_id":    exec.MemberID,
					"phase":        string(phase),
				}).Info("Execution cancelled by user")

				// Persist cancelled status
				if !e.config.SkipPersistence && e.store != nil {
					_ = e.store.UpdateStatus(ctx.Context, exec.ID, robottypes.ExecCancelled, "execution cancelled by user")
				}
				return exec, nil
			}

			// Normal failure case
			exec.Status = robottypes.ExecFailed
			exec.Error = err.Error()

			// Update UI field for failure with i18n
			failedPrefix := getLocalizedMessage(locale, "failed_prefix")
			phaseName := getLocalizedMessage(locale, "phase_"+string(phase))
			failureMsg := failedPrefix + phaseName
			e.updateUIFields(ctx, exec, "", failureMsg)

			kunlog.With(kunlog.F{
				"execution_id": exec.ID,
				"member_id":    exec.MemberID,
				"phase":        string(phase),
				"error":        err.Error(),
			}).Error("Phase execution failed: %v", err)
			// Persist failed status
			if !e.config.SkipPersistence && e.store != nil {
				_ = e.store.UpdateStatus(ctx.Context, exec.ID, robottypes.ExecFailed, err.Error())
			}
			return exec, nil
		}
	}

	// Mark completed
	exec.Status = robottypes.ExecCompleted
	now := time.Now()
	exec.EndTime = &now

	// Update UI field for completion with i18n
	e.updateUIFields(ctx, exec, "", getLocalizedMessage(locale, "completed"))

	duration := now.Sub(exec.StartTime)
	kunlog.With(kunlog.F{
		"execution_id": exec.ID,
		"member_id":    exec.MemberID,
		"duration_ms":  duration.Milliseconds(),
	}).Info("Execution completed successfully")

	// Persist completed status
	if !e.config.SkipPersistence && e.store != nil {
		if err := e.store.UpdateStatus(ctx.Context, exec.ID, robottypes.ExecCompleted, ""); err != nil {
			kunlog.With(kunlog.F{
				"execution_id": exec.ID,
				"error":        err,
			}).Warn("Failed to persist completed status: %v", err)
		}
	}

	event.Push(ctx.Context, robotevents.ExecCompleted, robotevents.ExecPayload{
		ExecutionID: exec.ID,
		MemberID:    exec.MemberID,
		TeamID:      exec.TeamID,
		Status:      string(robottypes.ExecCompleted),
		ChatID:      exec.ChatID,
	})

	return exec, nil
}

// runPhase executes a single phase
func (e *Executor) runPhase(ctx *robottypes.Context, exec *robottypes.Execution, phase robottypes.Phase, data interface{}, control robottypes.ExecutionControl) error {
	// Check if context is cancelled before starting this phase
	select {
	case <-ctx.Context.Done():
		return robottypes.ErrExecutionCancelled
	default:
	}

	// Wait if execution is paused (blocks until resumed or cancelled)
	if control != nil {
		if err := control.WaitIfPaused(); err != nil {
			return err // Returns ErrExecutionCancelled if cancelled while paused
		}
	}

	exec.Phase = phase

	kunlog.With(kunlog.F{
		"execution_id": exec.ID,
		"member_id":    exec.MemberID,
		"phase":        string(phase),
	}).Info("Phase started: %s", phase)

	// Persist phase change immediately (so frontend sees current phase)
	if !e.config.SkipPersistence && e.store != nil {
		if err := e.store.UpdatePhase(ctx.Context, exec.ID, phase, nil); err != nil {
			kunlog.With(kunlog.F{
				"execution_id": exec.ID,
				"phase":        string(phase),
				"error":        err,
			}).Warn("Failed to persist phase start: %v", err)
		}
	}

	if e.config.OnPhaseStart != nil {
		e.config.OnPhaseStart(phase)
	}

	phaseStart := time.Now()

	// Execute phase-specific logic
	var err error
	switch phase {
	case robottypes.PhaseInspiration:
		err = e.RunInspiration(ctx, exec, data)
	case robottypes.PhaseGoals:
		err = e.RunGoals(ctx, exec, data)
	case robottypes.PhaseTasks:
		err = e.RunTasks(ctx, exec, data)
	case robottypes.PhaseRun:
		err = e.RunExecution(ctx, exec, data)
	case robottypes.PhaseDelivery:
		err = e.RunDelivery(ctx, exec, data)
	case robottypes.PhaseLearning:
		err = e.RunLearning(ctx, exec, data)
	}

	if err != nil {
		if err == robottypes.ErrExecutionSuspended {
			kunlog.With(kunlog.F{
				"execution_id": exec.ID,
				"member_id":    exec.MemberID,
				"phase":        string(phase),
			}).Info("Phase suspended: %s (waiting for human input)", phase)
			return err
		}
		kunlog.With(kunlog.F{
			"execution_id": exec.ID,
			"member_id":    exec.MemberID,
			"phase":        string(phase),
			"error":        err.Error(),
		}).Error("Phase failed: %s - %v", phase, err)
		return err
	}

	// Persist phase output to database
	if !e.config.SkipPersistence && e.store != nil {
		phaseData := e.getPhaseData(exec, phase)
		if phaseData != nil {
			if err := e.store.UpdatePhase(ctx.Context, exec.ID, phase, phaseData); err != nil {
				// Log warning but don't fail execution
				kunlog.With(kunlog.F{
					"execution_id": exec.ID,
					"phase":        string(phase),
					"error":        err,
				}).Warn("Failed to persist phase %s data: %v", phase, err)
			}
		}
	}

	if e.config.OnPhaseEnd != nil {
		e.config.OnPhaseEnd(phase)
	}

	phaseDuration := time.Since(phaseStart).Milliseconds()
	kunlog.With(kunlog.F{
		"execution_id": exec.ID,
		"member_id":    exec.MemberID,
		"phase":        string(phase),
		"duration_ms":  phaseDuration,
	}).Info("Phase completed: %s (took %dms)", phase, phaseDuration)

	return nil
}

// getPhaseData extracts the output data for a specific phase from execution
func (e *Executor) getPhaseData(exec *robottypes.Execution, phase robottypes.Phase) interface{} {
	switch phase {
	case robottypes.PhaseInspiration:
		return exec.Inspiration
	case robottypes.PhaseGoals:
		return exec.Goals
	case robottypes.PhaseTasks:
		return exec.Tasks
	case robottypes.PhaseRun:
		return exec.Results
	case robottypes.PhaseDelivery:
		return exec.Delivery
	case robottypes.PhaseLearning:
		return exec.Learning
	default:
		return nil
	}
}

// ExecCount returns total execution count
func (e *Executor) ExecCount() int {
	return int(e.execCount.Load())
}

// CurrentCount returns currently running execution count
func (e *Executor) CurrentCount() int {
	return int(e.currentCount.Load())
}

// Reset resets the executor counters
func (e *Executor) Reset() {
	e.execCount.Store(0)
	e.currentCount.Store(0)
}

// DefaultStreamDelay is the simulated delay for Agent Stream calls
// This will be removed when real Agent calls are implemented
const DefaultStreamDelay = 50 * time.Millisecond

// simulateStreamDelay simulates the delay of an Agent Stream call
func (e *Executor) simulateStreamDelay() {
	time.Sleep(DefaultStreamDelay)
}

// initUIFields initializes UI display fields based on trigger type with i18n support
// Returns (name, currentTaskName)
func (e *Executor) initUIFields(trigger robottypes.TriggerType, input *robottypes.TriggerInput, robot *robottypes.Robot) (string, string) {
	// Determine locale for UI messages
	locale := getEffectiveLocale(robot, input)

	// Get localized default messages
	name := getLocalizedMessage(locale, "preparing")
	currentTaskName := getLocalizedMessage(locale, "starting")

	switch trigger {
	case robottypes.TriggerHuman:
		// For human trigger, extract name from first message
		if input != nil && len(input.Messages) > 0 {
			if content, ok := input.Messages[0].GetContentAsString(); ok && content != "" {
				// Use first 100 chars of message as name
				name = content
				if len(name) > 100 {
					name = name[:100] + "..."
				}
			}
		}
	case robottypes.TriggerClock:
		name = getLocalizedMessage(locale, "scheduled_execution")
	case robottypes.TriggerEvent:
		if input != nil && input.EventType != "" {
			name = getLocalizedMessage(locale, "event_prefix") + input.EventType
		} else {
			name = getLocalizedMessage(locale, "event_triggered")
		}
	}

	return name, currentTaskName
}

// getEffectiveLocale determines the locale for UI display
// Priority: input.Locale > robot.Config.DefaultLocale > "en"
func getEffectiveLocale(robot *robottypes.Robot, input *robottypes.TriggerInput) string {
	// 1. Human trigger with explicit locale
	if input != nil && input.Locale != "" {
		return input.Locale
	}
	// 2. Robot configured default
	if robot != nil && robot.Config != nil {
		return robot.Config.GetDefaultLocale()
	}
	// 3. System default
	return "en"
}

// i18n message maps for UI display fields
// Use simple locale codes (en, zh) as keys
var uiMessages = map[string]map[string]string{
	"en": {
		"preparing":           "Preparing...",
		"starting":            "Starting...",
		"scheduled_execution": "Scheduled execution",
		"event_prefix":        "Event: ",
		"event_triggered":     "Event triggered",
		"analyzing_context":   "Analyzing context...",
		"planning_goals":      "Planning goals...",
		"breaking_down_tasks": "Breaking down tasks...",
		"generating_delivery": "Generating delivery content...",
		"sending_delivery":    "Sending delivery...",
		"learning_from_exec":  "Learning from execution...",
		"completed":           "Completed",
		"cancelled":           "Cancelled",
		"failed_prefix":       "Failed at ",
		"task_prefix":         "Task",
		// Phase names for failure messages
		"phase_inspiration": "inspiration",
		"phase_goals":       "goals",
		"phase_tasks":       "tasks",
		"phase_run":         "execution",
		"phase_delivery":    "delivery",
		"phase_learning":    "learning",
	},
	"zh": {
		"preparing":           "准备中...",
		"starting":            "启动中...",
		"scheduled_execution": "定时执行",
		"event_prefix":        "事件: ",
		"event_triggered":     "事件触发",
		"analyzing_context":   "分析上下文...",
		"planning_goals":      "规划目标...",
		"breaking_down_tasks": "分解任务...",
		"generating_delivery": "生成交付内容...",
		"sending_delivery":    "正在发送...",
		"learning_from_exec":  "学习执行经验...",
		"completed":           "已完成",
		"cancelled":           "已取消",
		"failed_prefix":       "失败于",
		"task_prefix":         "任务",
		// Phase names for failure messages
		"phase_inspiration": "灵感阶段",
		"phase_goals":       "目标阶段",
		"phase_tasks":       "任务阶段",
		"phase_run":         "执行阶段",
		"phase_delivery":    "交付阶段",
		"phase_learning":    "学习阶段",
	},
}

// getLocalizedMessage returns a localized message for the given key
func getLocalizedMessage(locale string, key string) string {
	if messages, ok := uiMessages[locale]; ok {
		if msg, ok := messages[key]; ok {
			return msg
		}
	}
	// Fallback to English
	if messages, ok := uiMessages["en"]; ok {
		if msg, ok := messages[key]; ok {
			return msg
		}
	}
	return key // Return key as fallback
}

// updateUIFields updates UI display fields and persists to database
func (e *Executor) updateUIFields(ctx *robottypes.Context, exec *robottypes.Execution, name string, currentTaskName string) {
	// Update in-memory execution
	if name != "" {
		exec.Name = name
	}
	if currentTaskName != "" {
		exec.CurrentTaskName = currentTaskName
	}

	// Persist to database
	if !e.config.SkipPersistence && e.store != nil {
		if err := e.store.UpdateUIFields(ctx.Context, exec.ID, name, currentTaskName); err != nil {
			kunlog.With(kunlog.F{
				"execution_id": exec.ID,
				"error":        err,
			}).Warn("Failed to update UI fields: %v", err)
		}
	}
}

// updateTasksState persists the current tasks array with status to database
// This should be called after each task status change for real-time UI updates
func (e *Executor) updateTasksState(ctx *robottypes.Context, exec *robottypes.Execution) {
	if e.config.SkipPersistence || e.store == nil {
		return
	}

	// Convert Current to store.CurrentState
	var current *store.CurrentState
	if exec.Current != nil {
		current = &store.CurrentState{
			TaskIndex: exec.Current.TaskIndex,
			Progress:  exec.Current.Progress,
		}
	}

	if err := e.store.UpdateTasks(ctx.Context, exec.ID, exec.Tasks, current); err != nil {
		kunlog.With(kunlog.F{
			"execution_id": exec.ID,
			"error":        err,
		}).Warn("Failed to update tasks state: %v", err)
	}
}

// extractGoalName extracts the execution name from goals output
func extractGoalName(goals *robottypes.Goals) string {
	if goals == nil || goals.Content == "" {
		return ""
	}

	// Extract first non-empty, non-markdown-header line as the goal name
	content := goals.Content
	lines := strings.Split(content, "\n")

	for _, line := range lines {
		line = strings.TrimSpace(line)
		if line == "" {
			continue
		}
		// Skip markdown headers (# ## ### etc.)
		if strings.HasPrefix(line, "#") {
			continue
		}
		// Skip markdown horizontal rules (--- or ***)
		if strings.HasPrefix(line, "---") || strings.HasPrefix(line, "***") {
			continue
		}
		// Found a content line - strip markdown formatting
		line = stripMarkdownFormatting(line)
		// Limit length
		if len(line) > 150 {
			line = line[:150] + "..."
		}
		return line
	}

	// Fallback: if all lines are headers, use first header without # prefix
	for _, line := range lines {
		line = strings.TrimSpace(line)
		if line == "" {
			continue
		}
		// Strip leading # symbols
		line = strings.TrimLeft(line, "#")
		line = strings.TrimSpace(line)
		line = stripMarkdownFormatting(line)
		if line != "" {
			if len(line) > 150 {
				line = line[:150] + "..."
			}
			return line
		}
	}

	return ""
}

// stripMarkdownFormatting removes common markdown formatting from text
func stripMarkdownFormatting(s string) string {
	// Remove bold/italic markers
	s = strings.ReplaceAll(s, "**", "")
	s = strings.ReplaceAll(s, "__", "")
	s = strings.ReplaceAll(s, "*", "")
	s = strings.ReplaceAll(s, "_", "")
	// Remove inline code
	s = strings.ReplaceAll(s, "`", "")
	// Remove link syntax [text](url) -> text
	// Simple approach: just remove brackets and parentheses content
	for {
		start := strings.Index(s, "[")
		if start == -1 {
			break
		}
		end := strings.Index(s[start:], "]")
		if end == -1 {
			break
		}
		linkEnd := start + end
		// Check if followed by (url)
		if linkEnd+1 < len(s) && s[linkEnd+1] == '(' {
			parenEnd := strings.Index(s[linkEnd+1:], ")")
			if parenEnd != -1 {
				// Extract just the link text
				linkText := s[start+1 : linkEnd]
				s = s[:start] + linkText + s[linkEnd+1+parenEnd+1:]
				continue
			}
		}
		// Just remove brackets
		s = s[:start] + s[start+1:linkEnd] + s[linkEnd+1:]
	}
	return strings.TrimSpace(s)
}

// Suspend transitions the execution to waiting status, persists state, and returns
// ErrExecutionSuspended so the caller stops further phase processing.
func (e *Executor) Suspend(ctx *robottypes.Context, exec *robottypes.Execution, taskIndex int, question string) error {
	now := time.Now()
	taskID := ""
	if taskIndex >= 0 && taskIndex < len(exec.Tasks) {
		taskID = exec.Tasks[taskIndex].ID
		exec.Tasks[taskIndex].Status = robottypes.TaskWaitingInput
	}

	exec.Status = robottypes.ExecWaiting
	exec.WaitingTaskID = taskID
	exec.WaitingQuestion = question
	exec.WaitingSince = &now
	exec.ResumeContext = &robottypes.ResumeContext{
		TaskIndex:       taskIndex,
		PreviousResults: exec.Results,
	}

	if !e.config.SkipPersistence && e.store != nil {
		// Persist task state (waiting_input on the specific task)
		e.updateTasksState(ctx, exec)
		// Persist P3 results so UI can show completed tasks while waiting (§16.26)
		if err := e.store.UpdatePhase(ctx.Context, exec.ID, robottypes.PhaseRun, exec.Results); err != nil {
			kunlog.With(kunlog.F{
				"execution_id": exec.ID,
				"error":        err,
			}).Warn("Failed to persist partial results on suspend: %v", err)
		}
		// Persist suspend state atomically
		if err := e.store.UpdateSuspendState(ctx.Context, exec.ID, taskID, question, exec.ResumeContext); err != nil {
			kunlog.With(kunlog.F{
				"execution_id": exec.ID,
				"task_id":      taskID,
				"error":        err,
			}).Warn("Failed to persist suspend state: %v", err)
		}
	}

	kunlog.With(kunlog.F{
		"execution_id": exec.ID,
		"member_id":    exec.MemberID,
		"task_id":      taskID,
		"question":     question,
	}).Info("Execution suspended, waiting for human input")

	// Fire event (best-effort, errors are ignored)
	event.Push(ctx.Context, robotevents.ExecWaiting, robotevents.NeedInputPayload{
		ExecutionID: exec.ID,
		MemberID:    exec.MemberID,
		TeamID:      exec.TeamID,
		TaskID:      taskID,
		Question:    question,
		ChatID:      exec.ChatID,
	})

	return robottypes.ErrExecutionSuspended
}

// Resume resumes a suspended execution with human-provided input.
// Loads execution from DB, restores state, injects reply, and continues from the suspended task.
func (e *Executor) Resume(ctx *robottypes.Context, execID string, reply string) error {
	if ctx == nil {
		return fmt.Errorf("context is required for resume")
	}
	if execID == "" {
		return fmt.Errorf("execID cannot be empty")
	}
	if e.store == nil {
		return fmt.Errorf("store is required for resume")
	}

	// Load execution record from DB
	record, err := e.store.Get(ctx.Context, execID)
	if err != nil {
		return fmt.Errorf("failed to load execution: %w", err)
	}
	if record == nil {
		return fmt.Errorf("execution not found: %s", execID)
	}
	if record.Status != robottypes.ExecWaiting {
		return fmt.Errorf("execution %s is not in waiting status (current: %s)", execID, record.Status)
	}

	// Restore runtime execution from record
	exec := record.ToExecution()

	// Load robot from store
	if e.robotStore == nil {
		return fmt.Errorf("robot store is required for resume")
	}
	robotRecord, err := e.robotStore.Get(ctx.Context, exec.MemberID)
	if err != nil {
		return fmt.Errorf("failed to load robot: %w", err)
	}
	if robotRecord == nil {
		return fmt.Errorf("robot not found: %s", exec.MemberID)
	}
	robot, err := robotRecord.ToRobot()
	if err != nil {
		return fmt.Errorf("failed to convert robot record: %w", err)
	}
	exec.SetRobot(robot)

	// Re-add execution to robot's in-memory tracking (skips quota check per §16.30)
	robot.AddExecution(exec)

	// Maintain executor concurrency count (§16.21)
	e.currentCount.Add(1)
	defer e.currentCount.Add(-1)

	// Defer cleanup: mirror ExecuteWithControl's defer logic (§16.21)
	defer func() {
		if exec.Status == robottypes.ExecWaiting {
			return // re-suspended, keep tracking
		}
		robot.RemoveExecution(exec.ID)
		if robot.RunningCount() == 0 && !e.config.SkipPersistence && e.robotStore != nil {
			if err := e.robotStore.UpdateStatus(ctx.Context, robot.MemberID, robottypes.RobotIdle); err != nil {
				kunlog.With(kunlog.F{
					"member_id": robot.MemberID,
					"error":     err,
				}).Warn("Failed to update robot status to idle after resume: %v", err)
			}
		}
	}()

	// Handle __skip__: mark waiting task as skipped and advance to next task
	if reply == "__skip__" && exec.ResumeContext != nil {
		ti := exec.ResumeContext.TaskIndex
		if ti >= 0 && ti < len(exec.Tasks) {
			task := &exec.Tasks[ti]
			task.Status = robottypes.TaskSkipped
			exec.ResumeContext.PreviousResults = append(exec.ResumeContext.PreviousResults, robottypes.TaskResult{
				TaskID:   task.ID,
				Success:  false,
				Output:   "skipped",
				Duration: 0,
			})
			exec.ResumeContext.TaskIndex = ti + 1
			if !e.config.SkipPersistence && e.store != nil {
				e.updateTasksState(ctx, exec)
			}
		}
		reply = "" // Don't inject __skip__ as a message
	}

	// Inject reply into the waiting task's messages so the re-executed task gets context
	if exec.ResumeContext != nil {
		ti := exec.ResumeContext.TaskIndex
		if ti >= 0 && ti < len(exec.Tasks) && reply != "" {
			exec.Tasks[ti].Messages = append(exec.Tasks[ti].Messages, agentcontext.Message{
				Role:    agentcontext.RoleUser,
				Content: fmt.Sprintf("[Human reply] %s", reply),
			})
		}
	}

	// Clear waiting fields and transition back to running
	exec.Status = robottypes.ExecRunning
	exec.WaitingTaskID = ""
	exec.WaitingQuestion = ""
	exec.WaitingSince = nil

	if !e.config.SkipPersistence && e.store != nil {
		if err := e.store.UpdateResumeState(ctx.Context, exec.ID); err != nil {
			kunlog.With(kunlog.F{
				"execution_id": exec.ID,
				"error":        err,
			}).Warn("Failed to persist resume state: %v", err)
		}
	}

	kunlog.With(kunlog.F{
		"execution_id": exec.ID,
		"member_id":    exec.MemberID,
		"reply_len":    len(reply),
	}).Info("Execution resumed")

	event.Push(ctx.Context, robotevents.ExecResumed, robotevents.ExecPayload{
		ExecutionID: exec.ID,
		MemberID:    exec.MemberID,
		TeamID:      exec.TeamID,
		ChatID:      exec.ChatID,
	})

	// Continue P3 (Run) from where it was suspended
	if err := e.RunExecution(ctx, exec, nil); err != nil {
		if err == robottypes.ErrExecutionSuspended {
			return err
		}
		exec.Status = robottypes.ExecFailed
		exec.Error = err.Error()
		if !e.config.SkipPersistence && e.store != nil {
			_ = e.store.UpdateStatus(ctx.Context, exec.ID, robottypes.ExecFailed, err.Error())
		}
		return err
	}

	// Clear resume context after successful P3 completion
	exec.ResumeContext = nil

	// Continue with P4 (Delivery) and P5 (Learning)
	locale := getEffectiveLocale(robot, exec.Input)
	for _, phase := range []robottypes.Phase{robottypes.PhaseDelivery, robottypes.PhaseLearning} {
		if err := e.runPhase(ctx, exec, phase, nil, nil); err != nil {
			if err == robottypes.ErrExecutionSuspended {
				return err
			}
			exec.Status = robottypes.ExecFailed
			exec.Error = err.Error()
			failedPrefix := getLocalizedMessage(locale, "failed_prefix")
			phaseName := getLocalizedMessage(locale, "phase_"+string(phase))
			e.updateUIFields(ctx, exec, "", failedPrefix+phaseName)
			if !e.config.SkipPersistence && e.store != nil {
				_ = e.store.UpdateStatus(ctx.Context, exec.ID, robottypes.ExecFailed, err.Error())
			}
			return fmt.Errorf("resume phase %s failed: %w", phase, err)
		}
	}

	// Mark completed
	exec.Status = robottypes.ExecCompleted
	now := time.Now()
	exec.EndTime = &now
	e.updateUIFields(ctx, exec, "", getLocalizedMessage(locale, "completed"))
	if !e.config.SkipPersistence && e.store != nil {
		_ = e.store.UpdateStatus(ctx.Context, exec.ID, robottypes.ExecCompleted, "")
	}

	return nil
}

// Verify Executor implements types.Executor
var _ types.Executor = (*Executor)(nil)