opencode/yao
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge pull request #1437 from trheyi/main

Browse source 
Enhance Execution Control and Error Handling in Robot Manager
This commit is contained in:
Max 2026-01-28 20:13:51 +08:00 • committed by GitHub
commit e654ed4b06
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
17 changed files with 587 additions and 250 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

21
agent/robot/api/execution.go
View file

@ -130,7 +130,12 @@ func PauseExecution(ctx *types.Context, execID string) error {
return err
}
return mgr.PauseExecution(ctx, execID)
if err := mgr.PauseExecution(ctx, execID); err != nil {
return err
}
// Update database status to paused
return getExecutionStore().UpdateStatus(context.Background(), execID, types.ExecPaused, "")
}
// ResumeExecution resumes a paused execution
@ -144,7 +149,12 @@ func ResumeExecution(ctx *types.Context, execID string) error {
return err
}
return mgr.ResumeExecution(ctx, execID)
if err := mgr.ResumeExecution(ctx, execID); err != nil {
return err
}
// Update database status back to running
return getExecutionStore().UpdateStatus(context.Background(), execID, types.ExecRunning, "")
}
// StopExecution stops a running execution
@ -158,7 +168,12 @@ func StopExecution(ctx *types.Context, execID string) error {
return err
}
return mgr.StopExecution(ctx, execID)
if err := mgr.StopExecution(ctx, execID); err != nil {
return err
}
// Update database status to cancelled
return getExecutionStore().UpdateStatus(context.Background(), execID, types.ExecCancelled, "User cancelled")
}
// ==================== Execution Status API ====================

37
agent/robot/executor/dryrun/executor.go
View file

@ -42,8 +42,18 @@ func NewWithConfig(config types.DryRunConfig) *Executor {
}
}
// Execute simulates robot execution without real Agent calls
// Execute simulates robot execution without 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 simulates robot execution 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 simulates robot execution with execution control
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")
}
@ -54,9 +64,14 @@ func (e *Executor) Execute(ctx *robottypes.Context, robot *robottypes.Robot, tri
startPhaseIndex = 1 // Skip P0
}
// Use provided execID or generate new one
if execID == "" {
execID = fmt.Sprintf("dryrun_%d", time.Now().UnixNano())
}
// Create execution record
exec := &robottypes.Execution{
ID: fmt.Sprintf("dryrun_%d", time.Now().UnixNano()),
ID: execID,
MemberID: robot.MemberID,
TeamID: robot.TeamID,
TriggerType: trigger,
@ -106,6 +121,24 @@ func (e *Executor) Execute(ctx *robottypes.Context, robot *robottypes.Robot, tri
// Execute phases with mock data
phases := robottypes.AllPhases[startPhaseIndex:]
for _, phase := range phases {
// Check if cancelled
select {
case <-ctx.Context.Done():
exec.Status = robottypes.ExecCancelled
exec.Error = "execution cancelled"
return exec, nil
default:
}
// Wait if paused
if control != nil {
if err := control.WaitIfPaused(); err != nil {
exec.Status = robottypes.ExecCancelled
exec.Error = "execution cancelled while paused"
return exec, nil
}
}
exec.Phase = phase
// Phase start callback

30
agent/robot/executor/sandbox/executor.go
View file

@ -48,8 +48,18 @@ func NewWithConfig(config types.SandboxConfig) *Executor {
}
}
// Execute runs robot execution within sandbox constraints
// Execute runs robot execution within sandbox constraints (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 robot execution within sandbox constraints 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 robot execution within sandbox constraints with execution control
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")
}
@ -67,9 +77,14 @@ func (e *Executor) Execute(ctx *robottypes.Context, robot *robottypes.Robot, tri
startPhaseIndex = 1
}
// Use provided execID or generate new one
if execID == "" {
execID = fmt.Sprintf("sandbox_%d", time.Now().UnixNano())
}
// Create execution record
exec := &robottypes.Execution{
ID: fmt.Sprintf("sandbox_%d", time.Now().UnixNano()),
ID: execID,
MemberID: robot.MemberID,
TeamID: robot.TeamID,
TriggerType: trigger,
@ -99,7 +114,7 @@ func (e *Executor) Execute(ctx *robottypes.Context, robot *robottypes.Robot, tri
// Execute phases with sandbox constraints
phases := robottypes.AllPhases[startPhaseIndex:]
for _, phase := range phases {
// Check timeout
// Check timeout or cancellation
select {
case <-execCtx.Done():
exec.Status = robottypes.ExecFailed
@ -108,6 +123,15 @@ func (e *Executor) Execute(ctx *robottypes.Context, robot *robottypes.Robot, tri
default:
}
// Wait if paused
if control != nil {
if err := control.WaitIfPaused(); err != nil {
exec.Status = robottypes.ExecCancelled
exec.Error = "execution cancelled while paused"
return exec, nil
}
}
exec.Phase = phase
if e.config.OnPhaseStart != nil {

64
agent/robot/executor/standard/executor.go
View file

@ -45,8 +45,19 @@ func NewWithConfig(config types.Config) *Executor {
}
}
// Execute runs a robot through all applicable phases with real Agent calls
// 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")
}
@ -57,10 +68,15 @@ func (e *Executor) Execute(ctx *robottypes.Context, robot *robottypes.Robot, tri
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: utils.NewID(),
ID: execID,
MemberID: robot.MemberID,
TeamID: robot.TeamID,
TriggerType: trigger,
@ -174,7 +190,31 @@ func (e *Executor) Execute(ctx *robottypes.Context, robot *robottypes.Robot, tri
// Execute phases
phases := robottypes.AllPhases[startPhaseIndex:]
for _, phase := range phases {
if err := e.runPhase(ctx, exec, phase, data); err != nil {
if err := e.runPhase(ctx, exec, phase, data, control); err != nil {
// 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"))
log.With(log.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()
@ -228,7 +268,21 @@ func (e *Executor) Execute(ctx *robottypes.Context, robot *robottypes.Robot, tri
}
// runPhase executes a single phase
func (e *Executor) runPhase(ctx *robottypes.Context, exec *robottypes.Execution, phase robottypes.Phase, data interface{}) error {
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
log.With(log.F{
@ -422,6 +476,7 @@ var uiMessages = map[string]map[string]string{
"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
@ -445,6 +500,7 @@ var uiMessages = map[string]map[string]string{
"sending_delivery": "正在发送...",
"learning_from_exec": "学习执行经验...",
"completed": "已完成",
"cancelled": "已取消",
"failed_prefix": "失败于",
"task_prefix": "任务",
// Phase names for failure messages

12
agent/robot/executor/types/types.go
View file

@ -12,12 +12,22 @@ import (
// - DryRun: Plan-only mode, simulates execution without Agent calls
// - Sandbox: Isolated execution with resource limits and safety controls
type Executor interface {
// Execute runs a robot through all applicable phases
// ExecuteWithControl runs a robot through all applicable phases with execution control
// ctx: Execution context with auth and logging
// robot: Robot configuration and state
// trigger: What triggered this execution (clock, human, event)
// data: Trigger-specific data (human input, event payload, etc.)
// execID: Pre-generated execution ID (empty string to auto-generate)
// control: Optional execution control for pause/resume functionality
// Returns: Execution record with all phase outputs
ExecuteWithControl(ctx *robottypes.Context, robot *robottypes.Robot, trigger robottypes.TriggerType, data interface{}, execID string, control robottypes.ExecutionControl) (*robottypes.Execution, error)
// ExecuteWithID runs a robot through all applicable phases with a pre-generated execution ID
// This is a convenience wrapper around ExecuteWithControl without control
ExecuteWithID(ctx *robottypes.Context, robot *robottypes.Robot, trigger robottypes.TriggerType, data interface{}, execID string) (*robottypes.Execution, error)
// Execute runs a robot through all applicable phases (auto-generates execution ID)
// This is a convenience wrapper around ExecuteWithControl
Execute(ctx *robottypes.Context, robot *robottypes.Robot, trigger robottypes.TriggerType, data interface{}) (*robottypes.Execution, error)
// Metrics and control

28
agent/robot/manager/integration_concurrent_test.go
View file

@ -545,12 +545,22 @@ type trackingExecutor struct {
}
func (e *trackingExecutor) Execute(ctx *types.Context, robot *types.Robot, trigger types.TriggerType, data interface{}) (*types.Execution, error) {
return e.ExecuteWithControl(ctx, robot, trigger, data, "", nil)
}
func (e *trackingExecutor) ExecuteWithID(ctx *types.Context, robot *types.Robot, trigger types.TriggerType, data interface{}, execID string) (*types.Execution, error) {
return e.ExecuteWithControl(ctx, robot, trigger, data, execID, nil)
}
func (e *trackingExecutor) ExecuteWithControl(ctx *types.Context, robot *types.Robot, trigger types.TriggerType, data interface{}, execID string, control types.ExecutionControl) (*types.Execution, error) {
if robot == nil {
return nil, types.ErrRobotNotFound
}
// Use unique ID for each execution to properly track quota
execID := fmt.Sprintf("exec_%d", time.Now().UnixNano())
// Use provided execID or generate unique ID for each execution to properly track quota
if execID == "" {
execID = fmt.Sprintf("exec_%d", time.Now().UnixNano())
}
exec := &types.Execution{
ID: execID,
MemberID: robot.MemberID,
@ -604,12 +614,22 @@ type triggerTrackingExecutor struct {
}
func (e *triggerTrackingExecutor) Execute(ctx *types.Context, robot *types.Robot, trigger types.TriggerType, data interface{}) (*types.Execution, error) {
return e.ExecuteWithControl(ctx, robot, trigger, data, "", nil)
}
func (e *triggerTrackingExecutor) ExecuteWithID(ctx *types.Context, robot *types.Robot, trigger types.TriggerType, data interface{}, execID string) (*types.Execution, error) {
return e.ExecuteWithControl(ctx, robot, trigger, data, execID, nil)
}
func (e *triggerTrackingExecutor) ExecuteWithControl(ctx *types.Context, robot *types.Robot, trigger types.TriggerType, data interface{}, execID string, control types.ExecutionControl) (*types.Execution, error) {
if robot == nil {
return nil, types.ErrRobotNotFound
}
// Use unique ID for each execution to properly track quota
execID := fmt.Sprintf("exec_trigger_%s_%d", string(trigger), time.Now().UnixNano())
// Use provided execID or generate unique ID for each execution to properly track quota
if execID == "" {
execID = fmt.Sprintf("exec_trigger_%s_%d", string(trigger), time.Now().UnixNano())
}
exec := &types.Execution{
ID: execID,
MemberID: robot.MemberID,

14
agent/robot/manager/integration_control_test.go
View file

@ -495,12 +495,24 @@ type slowExecutor struct {
}
func (e *slowExecutor) Execute(ctx *types.Context, robot *types.Robot, trigger types.TriggerType, data interface{}) (*types.Execution, error) {
return e.ExecuteWithControl(ctx, robot, trigger, data, "", nil)
}
func (e *slowExecutor) ExecuteWithID(ctx *types.Context, robot *types.Robot, trigger types.TriggerType, data interface{}, execID string) (*types.Execution, error) {
return e.ExecuteWithControl(ctx, robot, trigger, data, execID, nil)
}
func (e *slowExecutor) ExecuteWithControl(ctx *types.Context, robot *types.Robot, trigger types.TriggerType, data interface{}, execID string, control types.ExecutionControl) (*types.Execution, error) {
if robot == nil {
return nil, types.ErrRobotNotFound
}
// Use provided execID or generate one
if execID == "" {
execID = "exec_slow_" + robot.MemberID
}
exec := &types.Execution{
ID: "exec_slow_" + robot.MemberID,
ID: execID,
MemberID: robot.MemberID,
TeamID: robot.TeamID,
TriggerType: trigger,

100
agent/robot/manager/manager.go
View file

@ -138,6 +138,16 @@ func (m *Manager) Start() error {
return fmt.Errorf("failed to load robots: %w", err)
}
// 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)
@ -250,17 +260,24 @@ func (m *Manager) Tick(parentCtx context.Context, now time.Time) error {
// continue
// }
// Create context with robot's own identity
// 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, robot.MemberID, robot.TeamID)
// Create context with robot's own identity and cancellable context
// Clock-triggered executions run as the robot itself
robotAuth := m.buildRobotAuth(robot)
ctx := types.NewContext(parentCtx, robotAuth)
execCtx := types.NewContext(ctrlExec.Context(), robotAuth)
// Create clock context for P0 inspiration
clockCtx := types.NewClockContext(now, robot.Config.Clock.TZ)
// Submit to pool
_, err := m.pool.Submit(ctx, robot, types.TriggerClock, clockCtx)
// Submit to pool with the cancellable context and execution control
_, err := m.pool.SubmitWithID(execCtx, robot, types.TriggerClock, clockCtx, execID, ctrlExec)
if err != nil {
// If submission failed, untrack the execution
m.execController.Untrack(execID)
// Log error but continue with other robots
// In production, this would be logged properly
continue
@ -408,9 +425,21 @@ func (m *Manager) TriggerManual(ctx *types.Context, memberID string, trigger typ
}
}
// Submit to pool
execID, err := m.pool.Submit(ctx, robot, trigger, data)
// 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)
// 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)
@ -579,17 +608,70 @@ func (m *Manager) HandleEvent(ctx *types.Context, req *types.EventRequest) (*typ
// PauseExecution pauses a running execution
func (m *Manager) PauseExecution(ctx *types.Context, execID string) error {
return m.execController.Pause(execID)
// 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 {
return m.execController.Resume(execID)
// 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 {
return m.execController.Stop(execID)
// 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

58
agent/robot/pool/pool.go
View file

@ -6,6 +6,7 @@ import (
"sync/atomic"
"github.com/yaoapp/yao/agent/robot/types"
"github.com/yaoapp/yao/agent/robot/utils"
)
// Default configuration values
@ -31,18 +32,23 @@ func DefaultConfig() *Config {
// ExecutorFactory creates an executor based on the mode
type ExecutorFactory func(mode types.ExecutorMode) types.Executor
// OnCompleteCallback is called when an execution completes (success or failure)
// Parameters: execID, memberID, status
type OnCompleteCallback func(execID, memberID string, status types.ExecStatus)
// Pool implements types.Pool interface
// Manages a pool of workers that execute robot jobs from a priority queue
type Pool struct {
size int // number of workers
queue *PriorityQueue // priority queue for pending jobs
executor types.Executor // default executor for running jobs
executorFactory ExecutorFactory // optional: factory for mode-specific executors
workers []*Worker // worker goroutines
running atomic.Int32 // number of currently running jobs
wg sync.WaitGroup // wait group for graceful shutdown
started bool // whether pool has been started
mu sync.RWMutex // protects started flag
size int // number of workers
queue *PriorityQueue // priority queue for pending jobs
executor types.Executor // default executor for running jobs
executorFactory ExecutorFactory // optional: factory for mode-specific executors
onComplete OnCompleteCallback // optional: callback when execution completes
workers []*Worker // worker goroutines
running atomic.Int32 // number of currently running jobs
wg sync.WaitGroup // wait group for graceful shutdown
started bool // whether pool has been started
mu sync.RWMutex // protects started flag
}
// New creates a new pool instance with default configuration
@ -85,6 +91,12 @@ func (p *Pool) SetExecutorFactory(factory ExecutorFactory) {
p.executorFactory = factory
}
// SetOnComplete sets the callback for execution completion
// Called when an execution finishes (completed, failed, or cancelled)
func (p *Pool) SetOnComplete(callback OnCompleteCallback) {
p.onComplete = callback
}
// GetExecutor returns the appropriate executor for the given mode
// If factory is set and mode is specified, uses factory; otherwise uses default
func (p *Pool) GetExecutor(mode types.ExecutorMode) types.Executor {
@ -149,10 +161,29 @@ func (p *Pool) Submit(ctx *types.Context, robot *types.Robot, trigger types.Trig
return p.SubmitWithMode(ctx, robot, trigger, data, "")
}
// GenerateExecID generates a new execution ID
// Exported so Manager can pre-generate IDs for tracking
func GenerateExecID() string {
return utils.NewID()
}
// SubmitWithMode submits a robot execution with specified executor mode
// executorMode: optional, overrides robot's config if provided
// Returns execution ID if successfully queued, error otherwise
// Note: This method does not support execution control (pause/resume)
func (p *Pool) SubmitWithMode(ctx *types.Context, robot *types.Robot, trigger types.TriggerType, data interface{}, executorMode types.ExecutorMode) (string, error) {
execID := GenerateExecID()
return p.submitWithIDAndMode(ctx, robot, trigger, data, execID, executorMode, nil)
}
// SubmitWithID submits a robot execution with a pre-generated execution ID
// This is used when the caller needs to track the execution before submission
func (p *Pool) SubmitWithID(ctx *types.Context, robot *types.Robot, trigger types.TriggerType, data interface{}, execID string, control types.ExecutionControl) (string, error) {
return p.submitWithIDAndMode(ctx, robot, trigger, data, execID, "", control)
}
// submitWithIDAndMode is the internal implementation that handles both cases
func (p *Pool) submitWithIDAndMode(ctx *types.Context, robot *types.Robot, trigger types.TriggerType, data interface{}, execID string, executorMode types.ExecutorMode, control types.ExecutionControl) (string, error) {
p.mu.RLock()
if !p.started {
p.mu.RUnlock()
@ -164,13 +195,15 @@ func (p *Pool) SubmitWithMode(ctx *types.Context, robot *types.Robot, trigger ty
return "", fmt.Errorf("robot cannot be nil")
}
// Create queue item
// Create queue item with the provided ID and control
item := &QueueItem{
Robot: robot,
Ctx: ctx,
Trigger: trigger,
Data: data,
ExecutorMode: executorMode,
ExecID: execID,
Control: control,
}
// Try to add to queue
@ -178,11 +211,6 @@ func (p *Pool) SubmitWithMode(ctx *types.Context, robot *types.Robot, trigger ty
return "", fmt.Errorf("queue full (max %d items)", p.queue.maxSize)
}
// Generate execution ID for tracking
// Note: Actual execution ID will be generated by Executor
// This is just a placeholder for the Submit return value
execID := fmt.Sprintf("queued_%s_%d", robot.MemberID, item.EnqueueTime.Unix())
return execID, nil
}

4
agent/robot/pool/queue.go
View file

@ -14,7 +14,9 @@ type QueueItem struct {
Ctx *types.Context
Trigger types.TriggerType
Data interface{}
ExecutorMode types.ExecutorMode // optional: override robot's executor mode
ExecutorMode types.ExecutorMode // optional: override robot's executor mode
ExecID string // pre-generated execution ID for tracking
Control types.ExecutionControl // execution control for pause/resume/stop
EnqueueTime time.Time
Priority int // calculated priority for sorting
Index int // index in heap (managed by container/heap)

20
agent/robot/pool/worker.go
View file

@ -79,9 +79,10 @@ func (w *Worker) execute(item *QueueItem) {
// Get executor based on mode (uses factory if available, otherwise default)
exec := w.pool.GetExecutor(item.ExecutorMode)
// Execute via Executor interface
// Note: Executor.Execute() does atomic quota check via TryAcquireSlot()
execution, err := exec.Execute(item.Ctx, item.Robot, item.Trigger, item.Data)
// Execute via Executor interface with pre-generated ID and control
// Note: Executor.ExecuteWithControl() does atomic quota check via TryAcquireSlot()
// The control parameter allows executor to check pause state during execution
execution, err := exec.ExecuteWithControl(item.Ctx, item.Robot, item.Trigger, item.Data, item.ExecID, item.Control)
if err != nil {
// Check if it's a quota error (race condition - another worker got the slot)
@ -91,12 +92,25 @@ func (w *Worker) execute(item *QueueItem) {
}
fmt.Printf("Worker %d: Execution failed for robot %s: %v\n",
w.id, item.Robot.MemberID, err)
// Notify completion callback with appropriate status
if w.pool.onComplete != nil {
// Determine status based on error type
status := types.ExecFailed
if err == types.ErrExecutionCancelled {
status = types.ExecCancelled
}
w.pool.onComplete(item.ExecID, item.Robot.MemberID, status)
}
return
}
if execution != nil {
fmt.Printf("Worker %d: Execution %s completed for robot %s (status: %s)\n",
w.id, execution.ID, item.Robot.MemberID, execution.Status)
// Notify completion callback
if w.pool.onComplete != nil {
w.pool.onComplete(execution.ID, item.Robot.MemberID, execution.Status)
}
}
}

1
agent/robot/types/enums.go
View file

@ -46,6 +46,7 @@ type ExecStatus string
const (
ExecPending ExecStatus = "pending"
ExecRunning ExecStatus = "running"
ExecPaused ExecStatus = "paused"
ExecCompleted ExecStatus = "completed"
ExecFailed ExecStatus = "failed"
ExecCancelled ExecStatus = "cancelled"

1
agent/robot/types/enums_test.go
View file

@ -41,6 +41,7 @@ func TestTriggerTypeEnum(t *testing.T) {
func TestExecStatusEnum(t *testing.T) {
assert.Equal(t, types.ExecStatus("pending"), types.ExecPending)
assert.Equal(t, types.ExecStatus("running"), types.ExecRunning)
assert.Equal(t, types.ExecStatus("paused"), types.ExecPaused)
assert.Equal(t, types.ExecStatus("completed"), types.ExecCompleted)
assert.Equal(t, types.ExecStatus("failed"), types.ExecFailed)
assert.Equal(t, types.ExecStatus("cancelled"), types.ExecCancelled)

21
agent/robot/types/interfaces.go
View file

@ -7,6 +7,19 @@ import "time"
// External API is defined in api/api.go
// All interfaces use *Context (not context.Context) for consistency.
// ExecutionControl provides pause/resume/stop control for running executions
// This interface is implemented by trigger.ControlledExecution
type ExecutionControl interface {
// IsPaused returns true if execution is paused
IsPaused() bool
// IsCancelled returns true if execution is cancelled
IsCancelled() bool
// WaitIfPaused blocks until resumed or cancelled, returns error if cancelled
WaitIfPaused() error
// CheckCancelled returns ErrExecutionCancelled if cancelled
CheckCancelled() error
}
// Manager - robot lifecycle and clock trigger management
type Manager interface {
Start() error
@ -16,6 +29,14 @@ type Manager interface {
// Executor - executes robot phases
type Executor interface {
// ExecuteWithControl runs execution with pre-generated ID and execution control (used by pool)
// control: optional, allows pause/resume functionality
ExecuteWithControl(ctx *Context, robot *Robot, trigger TriggerType, data interface{}, execID string, control ExecutionControl) (*Execution, error)
// ExecuteWithID runs execution with a pre-generated ID but no control (for backward compatibility)
ExecuteWithID(ctx *Context, robot *Robot, trigger TriggerType, data interface{}, execID string) (*Execution, error)
// Execute runs execution with auto-generated ID (for direct calls)
Execute(ctx *Context, robot *Robot, trigger TriggerType, data interface{}) (*Execution, error)
// Metrics and control (for monitoring and testing)

338
data/bindata.go
View file

File diff suppressed because it is too large Load diff

5
openapi/agent/robot/execution.go
View file

@ -2,6 +2,7 @@ package robot
import (
"errors"
"strings"
"github.com/gin-gonic/gin"
"github.com/yaoapp/kun/log"
@ -320,10 +321,10 @@ func handleExecutionControl(c *gin.Context, action string) {
// Check for common errors
errMsg := controlErr.Error()
if errMsg == "execution_id is required" || errMsg == "execution not found" {
if errMsg == "execution_id is required" || strings.Contains(errMsg, "execution not found") {
errorResp := &response.ErrorResponse{
Code: response.ErrInvalidRequest.Code,
ErrorDescription: "Execution not found: " + execID,
ErrorDescription: "Execution not found or not running: " + execID,
}
response.RespondWithError(c, response.StatusNotFound, errorResp)
return

83
yao/models/agent/execution.mod.yao
View file

@ -6,13 +6,16 @@
"builtin": true,
"readonly": false,
"sort": 9999,
"table": { "name": "agent_execution", "comment": "Robot execution history table" },
"table": {
"name": "agent_execution",
"comment": "Robot execution history table",
},
"columns": [
{
"name": "id",
"type": "ID",
"label": "ID",
"comment": "Auto-increment primary key"
"comment": "Auto-increment primary key",
},
{
"name": "execution_id",
@ -22,7 +25,7 @@
"length": 64,
"nullable": false,
"unique": true,
"index": true
"index": true,
},
{
"name": "member_id",
@ -31,7 +34,7 @@
"comment": "Robot member ID (user identity from __yao.member)",
"length": 64,
"nullable": false,
"index": true
"index": true,
},
{
"name": "team_id",
@ -40,7 +43,7 @@
"comment": "Team ID the robot belongs to",
"length": 64,
"nullable": false,
"index": true
"index": true,
},
{
"name": "job_id",
@ -49,7 +52,7 @@
"comment": "Linked job.Job ID for monitoring",
"length": 64,
"nullable": true,
"index": true
"index": true,
},
{
"name": "trigger_type",
@ -58,41 +61,55 @@
"comment": "How this execution was triggered",
"option": ["clock", "human", "event"],
"nullable": false,
"index": true
"index": true,
},
{
"name": "status",
"type": "enum",
"label": "Status",
"comment": "Execution status",
"option": ["pending", "running", "completed", "failed", "cancelled"],
"option": [
"pending",
"running",
"paused",
"completed",
"failed",
"cancelled",
],
"default": "pending",
"nullable": false,
"index": true
"index": true,
},
{
"name": "phase",
"type": "enum",
"label": "Phase",
"comment": "Current execution phase",
"option": ["inspiration", "goals", "tasks", "run", "delivery", "learning"],
"option": [
"inspiration",
"goals",
"tasks",
"run",
"delivery",
"learning",
],
"default": "inspiration",
"nullable": false,
"index": true
"index": true,
},
{
"name": "current",
"type": "json",
"label": "Current State",
"comment": "Current executing state (task_index, progress)",
"nullable": true
"nullable": true,
},
{
"name": "error",
"type": "text",
"label": "Error",
"comment": "Error message if execution failed",
"nullable": true
"nullable": true,
},
{
"name": "name",
@ -100,7 +117,7 @@
"label": "Name",
"comment": "Execution title for UI display (updated by executor at goals phase)",
"length": 512,
"nullable": true
"nullable": true,
},
{
"name": "current_task_name",
@ -108,56 +125,56 @@
"label": "Current Task Name",
"comment": "Current task description for UI display (updated by executor at run phase)",
"length": 512,
"nullable": true
"nullable": true,
},
{
"name": "input",
"type": "json",
"label": "Input",
"comment": "Original trigger input (TriggerInput)",
"nullable": true
"nullable": true,
},
{
"name": "inspiration",
"type": "json",
"label": "Inspiration",
"comment": "P0 output (InspirationReport)",
"nullable": true
"nullable": true,
},
{
"name": "goals",
"type": "json",
"label": "Goals",
"comment": "P1 output (Goals)",
"nullable": true
"nullable": true,
},
{
"name": "tasks",
"type": "json",
"label": "Tasks",
"comment": "P2 output ([]Task)",
"nullable": true
"nullable": true,
},
{
"name": "results",
"type": "json",
"label": "Results",
"comment": "P3 output ([]TaskResult)",
"nullable": true
"nullable": true,
},
{
"name": "delivery",
"type": "json",
"label": "Delivery",
"comment": "P4 output (DeliveryResult)",
"nullable": true
"nullable": true,
},
{
"name": "learning",
"type": "json",
"label": "Learning",
"comment": "P5 output ([]LearningEntry)",
"nullable": true
"nullable": true,
},
{
"name": "start_time",
@ -165,7 +182,7 @@
"label": "Start Time",
"comment": "Execution start timestamp",
"nullable": true,
"index": true
"index": true,
},
{
"name": "end_time",
@ -173,42 +190,42 @@
"label": "End Time",
"comment": "Execution end timestamp",
"nullable": true,
"index": true
}
"index": true,
},
],
"relations": {
"member": {
"type": "hasOne",
"model": "__yao.member",
"key": "member_id",
"foreign": "member_id"
}
"foreign": "member_id",
},
},
"indexes": [
{
"name": "idx_agent_execution_member_status",
"columns": ["member_id", "status"],
"type": "index",
"comment": "Index for member execution queries with status filter"
"comment": "Index for member execution queries with status filter",
},
{
"name": "idx_agent_execution_team_status",
"columns": ["team_id", "status"],
"type": "index",
"comment": "Index for team execution queries with status filter"
"comment": "Index for team execution queries with status filter",
},
{
"name": "idx_agent_execution_trigger_start",
"columns": ["trigger_type", "start_time"],
"type": "index",
"comment": "Index for trigger type analysis"
"comment": "Index for trigger type analysis",
},
{
"name": "idx_agent_execution_member_start",
"columns": ["member_id", "start_time"],
"type": "index",
"comment": "Index for robot execution history by member"
}
"comment": "Index for robot execution history by member",
},
],
"option": { "timestamps": true, "soft_deletes": false, "permission": true }
"option": { "timestamps": true, "soft_deletes": false, "permission": true },
}