Merge pull request #1132 from trheyi/main

Enhance job execution and management features

job/data.go

@@ -103,6 +103,14 @@ func SaveJob(job *Job) error {
 		return fmt.Errorf("job model not found")
 	}
 
+	// Ensure category exists before saving job
+	if job.CategoryID != "" {
+		_, err := ensureCategoryExists(job.CategoryID)
+		if err != nil {
+			return fmt.Errorf("failed to ensure category exists: %w", err)
+		}
+	}
+
 	data := structToMap(job)
 	now := time.Now()
 
@@ -377,6 +385,64 @@ func GetOrCreateCategory(name, description string) (*Category, error) {
 	return category, nil
 }
 
+// ensureCategoryExists ensures a category exists by category_id, creates default if needed
+func ensureCategoryExists(categoryID string) (*Category, error) {
+	mod := model.Select("__yao.job.category")
+	if mod == nil {
+		return nil, fmt.Errorf("job category model not found")
+	}
+
+	// Try to find existing category by category_id
+	param := model.QueryParam{
+		Wheres: []model.QueryWhere{
+			{Column: "category_id", Value: categoryID},
+		},
+		Limit: 1,
+	}
+
+	results, err := mod.Get(param)
+	if err != nil {
+		return nil, err
+	}
+
+	if len(results) > 0 {
+		// Category exists
+		category := &Category{}
+		if err := mapToStruct(results[0], category); err != nil {
+			return nil, err
+		}
+		return category, nil
+	}
+
+	// Create default category if it doesn't exist
+	var categoryName, categoryDesc string
+	if categoryID == "default" {
+		categoryName = "Default"
+		categoryDesc = "Default job category"
+	} else {
+		categoryName = categoryID
+		categoryDesc = fmt.Sprintf("Auto-created category: %s", categoryID)
+	}
+
+	category := &Category{
+		CategoryID:  categoryID,
+		Name:        categoryName,
+		Description: &categoryDesc,
+		Sort:        0,
+		System:      categoryID == "default", // Mark default as system category
+		Enabled:     true,
+		Readonly:    false,
+		CreatedAt:   time.Now(),
+		UpdatedAt:   time.Now(),
+	}
+
+	if err := SaveCategory(category); err != nil {
+		return nil, err
+	}
+
+	return category, nil
+}
+
 // ========================
 // Logs methods
 // ========================
@@ -481,6 +547,15 @@ func GetExecutions(jobID string) ([]*Execution, error) {
 		if err := mapToStruct(result, execution); err != nil {
 			continue
 		}
+
+		// Restore ExecutionConfig from ConfigSnapshot if available
+		if execution.ConfigSnapshot != nil && len(*execution.ConfigSnapshot) > 0 {
+			var config ExecutionConfig
+			if err := jsoniter.Unmarshal(*execution.ConfigSnapshot, &config); err == nil {
+				execution.ExecutionConfig = &config
+			}
+		}
+
 		executions = append(executions, execution)
 	}
 
@@ -576,6 +651,14 @@ func GetExecution(executionID string, param model.QueryParam) (*Execution, error
 		return nil, err
 	}
 
+	// Restore ExecutionConfig from ConfigSnapshot if available
+	if execution.ConfigSnapshot != nil && len(*execution.ConfigSnapshot) > 0 {
+		var config ExecutionConfig
+		if err := jsoniter.Unmarshal(*execution.ConfigSnapshot, &config); err == nil {
+			execution.ExecutionConfig = &config
+		}
+	}
+
 	return execution, nil
 }
 
@@ -626,6 +709,11 @@ func SaveExecution(execution *Execution) error {
 		}
 	}
 
+	// Update related Job information after execution changes
+	if err := updateJobProgress(execution.JobID); err != nil {
+		return fmt.Errorf("failed to update job progress: %w", err)
+	}
+
 	return nil
 }
 
@@ -723,3 +811,85 @@ func mapToStruct(m maps.MapStr, v interface{}) error {
 	}
 	return jsoniter.Unmarshal(data, v)
 }
+
+// updateJobProgress updates job progress and status based on its executions
+func updateJobProgress(jobID string) error {
+	// Skip if jobID is empty
+	if jobID == "" {
+		return nil
+	}
+
+	// Get all executions for this job
+	executions, err := GetExecutions(jobID)
+	if err != nil {
+		return fmt.Errorf("failed to get executions for job %s: %w", jobID, err)
+	}
+
+	if len(executions) == 0 {
+		return nil // No executions to process
+	}
+
+	// Calculate overall job progress and status
+	totalExecutions := len(executions)
+	completedCount := 0
+	failedCount := 0
+	runningCount := 0
+	totalProgress := 0
+
+	for _, execution := range executions {
+		totalProgress += execution.Progress
+
+		switch execution.Status {
+		case "completed":
+			completedCount++
+		case "failed":
+			failedCount++
+		case "running":
+			runningCount++
+		}
+	}
+
+	// Calculate average progress
+	averageProgress := totalProgress / totalExecutions
+
+	// Determine job status
+	var jobStatus string
+	if completedCount == totalExecutions {
+		jobStatus = "completed"
+	} else if failedCount > 0 && runningCount == 0 && completedCount+failedCount == totalExecutions {
+		jobStatus = "failed"
+	} else if runningCount > 0 || completedCount > 0 {
+		jobStatus = "running"
+	} else {
+		jobStatus = "ready" // All executions are queued
+	}
+
+	// Update job in database
+	jobMod := model.Select("__yao.job")
+	if jobMod == nil {
+		return fmt.Errorf("job model not found")
+	}
+
+	updateData := map[string]interface{}{
+		"status":     jobStatus,
+		"updated_at": time.Now(),
+	}
+
+	// Add progress field if Job model supports it
+	// Note: This assumes Job model has a progress field, you may need to add it to the schema
+	updateData["progress"] = averageProgress
+
+	param := model.QueryParam{
+		Wheres: []model.QueryWhere{
+			{Column: "job_id", Value: jobID},
+		},
+		Limit: 1,
+	}
+
+	_, err = jobMod.UpdateWhere(param, updateData)
+	if err != nil {
+		return fmt.Errorf("failed to update job progress: %w", err)
+	}
+
+	return nil
+}

job/data_test.go

@@ -22,9 +22,10 @@ func TestJobCRUD(t *testing.T) {
 		t.Fatalf("Failed to create test category: %v", err)
 	}
 
-	// Test job creation
+	// Test job creation with unique ID
+	timestamp := time.Now().UnixNano()
 	testJob := &job.Job{
-		JobID:         "test-job-crud-001",
+		JobID:         fmt.Sprintf("test-job-crud-%d", timestamp),
 		Name:          "Test CRUD Job",
 		CategoryID:    category.CategoryID,
 		Status:        "draft",
@@ -41,6 +42,12 @@ func TestJobCRUD(t *testing.T) {
 		UpdatedAt:     time.Now(),
 	}
 	
+	// Ensure cleanup even if test fails
+	defer func() {
+		job.RemoveJobs([]string{testJob.JobID})
+		job.RemoveCategories([]string{category.CategoryID})
+	}()
+
 	// Test SaveJob (Create)
 	err = job.SaveJob(testJob)
 	if err != nil {
@@ -136,9 +143,10 @@ func TestCategoryCRUD(t *testing.T) {
 	test.Prepare(t, config.Conf)
 	defer test.Clean()
 
-	// Test category creation
+	// Test category creation with unique ID
+	timestamp := time.Now().UnixNano()
 	testCategory := &job.Category{
-		CategoryID:  "test-category-crud-001",
+		CategoryID:  fmt.Sprintf("test-category-crud-%d", timestamp),
 		Name:        "Test CRUD Category",
 		Description: stringPtr("Test category for CRUD operations"),
 		Sort:        1,
@@ -149,6 +157,11 @@ func TestCategoryCRUD(t *testing.T) {
 		UpdatedAt:   time.Now(),
 	}
 	
+	// Ensure cleanup even if test fails
+	defer func() {
+		job.RemoveCategories([]string{testCategory.CategoryID})
+	}()
+
 	// Test SaveCategory (Create)
 	err := job.SaveCategory(testCategory)
 	if err != nil {
@@ -249,9 +262,10 @@ func TestExecutionCRUD(t *testing.T) {
 	test.Prepare(t, config.Conf)
 	defer test.Clean()
 
-	// Create test job first
+	// Create test job first with unique ID
+	timestamp := time.Now().UnixNano()
 	testJob := &job.Job{
-		JobID:         "test-execution-job-001",
+		JobID:         fmt.Sprintf("test-execution-job-%d", timestamp),
 		Name:          "Test Execution Job",
 		CategoryID:    "default",
 		Status:        "ready",
@@ -268,9 +282,15 @@ func TestExecutionCRUD(t *testing.T) {
 		t.Fatalf("Failed to create test job: %v", err)
 	}
 
-	// Test execution creation
+	// Ensure cleanup even if test fails
+	defer func() {
+		job.RemoveJobs([]string{testJob.JobID})
+	}()
+
+	// Test execution creation with unique ID
+	executionTimestamp := time.Now().UnixNano() + 1 // Ensure different from job timestamp
 	testExecution := &job.Execution{
-		ExecutionID:     "test-execution-crud-001",
+		ExecutionID:     fmt.Sprintf("test-execution-crud-%d", executionTimestamp),
 		JobID:           testJob.JobID,
 		Status:          "queued",
 		TriggerCategory: "manual",
@@ -359,9 +379,10 @@ func TestLogCRUD(t *testing.T) {
 	test.Prepare(t, config.Conf)
 	defer test.Clean()
 
-	// Create test job first
+	// Create test job first with unique ID
+	timestamp := time.Now().UnixNano()
 	testJob := &job.Job{
-		JobID:         "test-log-job-001",
+		JobID:         fmt.Sprintf("test-log-job-%d", timestamp),
 		Name:          "Test Log Job",
 		CategoryID:    "default",
 		Status:        "ready",
@@ -378,6 +399,11 @@ func TestLogCRUD(t *testing.T) {
 		t.Fatalf("Failed to create test job: %v", err)
 	}
 	
+	// Ensure cleanup even if test fails
+	defer func() {
+		job.RemoveJobs([]string{testJob.JobID})
+	}()
+
 	// Test log creation
 	testLog := &job.Log{
 		JobID:     testJob.JobID,

job/execution.go

@@ -1,70 +1,70 @@
 package job
 
 import (
+	"encoding/json"
 	"fmt"
-	"sync"
 	"time"
 
+	jsoniter "github.com/json-iterator/go"
 	"github.com/yaoapp/gou/model"
 	"github.com/yaoapp/kun/log"
 )
 
-// handlerRegistry stores registered handlers for jobs
+// Add adds a new execution with Yao process (default execution type)
-var handlerRegistry = make(map[string]HandlerFunc)
+func (j *Job) Add(options *ExecutionOptions, processName string, args ...interface{}) error {
-var handlerRegistryMutex sync.RWMutex
+	return j.addExecution(options, &ExecutionConfig{
-
+		Type:        ExecutionTypeProcess,
-// SetHandler sets a handler for a job ID (for testing)
+		ProcessName: processName,
-func SetHandler(jobID string, handler HandlerFunc) {
+		ProcessArgs: args,
-	handlerRegistryMutex.Lock()
+	})
-	defer handlerRegistryMutex.Unlock()
-	handlerRegistry[jobID] = handler
 }
 
-// getHandler gets a handler for a job ID (thread-safe)
+// AddCommand adds a new execution with system command
-func getHandler(jobID string) (HandlerFunc, bool) {
+func (j *Job) AddCommand(options *ExecutionOptions, command string, args []string, env map[string]string) error {
-	handlerRegistryMutex.RLock()
+	return j.addExecution(options, &ExecutionConfig{
-	defer handlerRegistryMutex.RUnlock()
+		Type:        ExecutionTypeCommand,
-	handler, exists := handlerRegistry[jobID]
+		Command:     command,
-	return handler, exists
+		CommandArgs: args,
+		Environment: env,
+	})
 }
 
-// Add add a new execution to the job with handler
+// addExecution is the internal method to create execution records
-func (j *Job) Add(priority int, handler HandlerFunc) error {
+func (j *Job) addExecution(options *ExecutionOptions, config *ExecutionConfig) error {
-	// Set job priority
+	// Set default options if nil
-	j.Priority = priority
+	if options == nil {
-
+		options = &ExecutionOptions{
-	// Auto-create or get category if not set
+			Priority:   0,
-	if j.CategoryID == "" {
+			SharedData: make(map[string]interface{}),
-		category, err := GetOrCreateCategory("default", "Default job category")
-		if err != nil {
-			log.Warn("Failed to create default category: %v", err)
-			j.CategoryID = "default"
-		} else {
-			j.CategoryID = category.CategoryID
 		}
 	}
 
-	// Set default values
+	// Serialize ExecutionConfig to JSON for ConfigSnapshot
-	if j.Status == "" {
+	configBytes, err := jsoniter.Marshal(config)
-		j.Status = "draft"
-	}
-	if j.MaxWorkerNums == 0 {
-		j.MaxWorkerNums = 1
-	}
-	if j.CreatedBy == "" {
-		j.CreatedBy = "system"
-	}
-
-	// Save job to database first
-	err := SaveJob(j)
 	if err != nil {
-		return err
+		return fmt.Errorf("failed to serialize execution config: %w", err)
+	}
+	configSnapshot := json.RawMessage(configBytes)
+
+	// Create new execution record with options and config
+	execution := &Execution{
+		ExecutionID:      "", // Will be generated in SaveExecution
+		JobID:            j.JobID,
+		Status:           "queued",
+		TriggerCategory:  "manual",
+		RetryAttempt:     0,
+		Progress:         0,
+		ExecutionConfig:  config,          // Keep in memory for runtime use
+		ConfigSnapshot:   &configSnapshot, // Store in database
+		ExecutionOptions: options,
+		CreatedAt:        time.Now(),
+		UpdatedAt:        time.Now(),
 	}
 
-	// Store handler in registry using the final JobID (thread-safe)
+	// Save execution to database
-	handlerRegistryMutex.Lock()
+	if err := SaveExecution(execution); err != nil {
-	handlerRegistry[j.JobID] = handler
+		return fmt.Errorf("failed to create execution record: %w", err)
-	handlerRegistryMutex.Unlock()
+	}
 
 	return nil
 }

job/goroutine.go

@@ -1,4 +1,263 @@
 package job
 
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+	"os/exec"
+
+	jsoniter "github.com/json-iterator/go"
+	"github.com/yaoapp/gou/model"
+	"github.com/yaoapp/gou/process"
+)
+
 // Goroutine the goroutine mode
 type Goroutine struct{}
+
+// ExecuteYaoProcess executes a Yao process using goroutine mode (process API)
+func (g *Goroutine) ExecuteYaoProcess(ctx context.Context, work *WorkRequest, progress *Progress) error {
+	config := work.Execution.ExecutionConfig
+
+	work.Execution.Info("Executing Yao process: %s (goroutine mode)", config.ProcessName)
+
+	// Create process with context
+	proc := process.NewWithContext(ctx, config.ProcessName, config.ProcessArgs...)
+
+	// Set shared data from ExecutionOptions to process context
+	if work.Execution.ExecutionOptions != nil && work.Execution.ExecutionOptions.SharedData != nil {
+		// SharedData itself is the Global context
+		proc.WithGlobal(work.Execution.ExecutionOptions.SharedData)
+
+		// Check if there's a 'sid' field in SharedData for session context
+		if sidValue, exists := work.Execution.ExecutionOptions.SharedData["sid"]; exists {
+			if sid, ok := sidValue.(string); ok {
+				proc.WithSID(sid)
+			}
+		}
+	}
+
+	// Set callback function to handle real-time progress updates
+	proc.WithCallback(func(process *process.Process, data map[string]interface{}) error {
+		if data == nil {
+			return nil
+		}
+
+		// Check if this is a progress update
+		if dataType, ok := data["type"].(string); ok && dataType == "progress" {
+			// Extract progress and message using helper function
+			progressInt, message := extractProgressData(data)
+
+			// Update execution progress
+			if progressInt >= 0 {
+				work.Execution.Progress = progressInt
+			}
+
+			// Log progress message
+			if message != "" {
+				work.Execution.Info("Progress update: %s (%.1f%%)", message, float64(work.Execution.Progress))
+			}
+
+			// Update progress tracker using Set method
+			if progress != nil && (progressInt >= 0 || message != "") {
+				progress.Set(progressInt, message)
+			}
+
+			// Save progress update to database
+			if saveErr := SaveExecution(work.Execution); saveErr != nil {
+				work.Execution.Error("Failed to save progress update: %s", saveErr.Error())
+				return saveErr
+			}
+		}
+
+		return nil
+	})
+
+	// Execute the process
+	err := proc.Execute()
+
+	// Always get result and release resources, even if there was an error
+	result := proc.Value()
+	proc.Release()
+
+	if err != nil {
+		work.Execution.Error("Yao process failed: %s", err.Error())
+
+		// Update execution with error info
+		work.Execution.Status = "failed"
+		if saveErr := SaveExecution(work.Execution); saveErr != nil {
+			work.Execution.Error("Failed to save execution error: %s", saveErr.Error())
+		}
+
+		return fmt.Errorf("yao process execution failed: %w", err)
+	}
+
+	work.Execution.Info("Yao process completed successfully, result: %v", result)
+
+	// Update execution with success result
+	work.Execution.Status = "completed"
+	work.Execution.Progress = 100
+	if result != nil {
+		if resultBytes, err := jsoniter.Marshal(result); err == nil {
+			work.Execution.Result = (*json.RawMessage)(&resultBytes)
+		}
+	}
+
+	if saveErr := SaveExecution(work.Execution); saveErr != nil {
+		work.Execution.Error("Failed to save execution result: %s", saveErr.Error())
+		return fmt.Errorf("failed to save execution result: %w", saveErr)
+	}
+
+	return nil
+}
+
+// ExecuteSystemCommand executes a system command using goroutine mode
+func (g *Goroutine) ExecuteSystemCommand(ctx context.Context, work *WorkRequest, progress *Progress) error {
+	config := work.Execution.ExecutionConfig
+
+	work.Execution.Info("Executing command: %s (goroutine mode)", config.Command)
+
+	// Create command with context for cancellation support
+	cmd := exec.CommandContext(ctx, config.Command, config.CommandArgs...)
+
+	// Set environment variables if provided
+	if len(config.Environment) > 0 {
+		env := os.Environ()
+		for key, value := range config.Environment {
+			env = append(env, fmt.Sprintf("%s=%s", key, value))
+		}
+		cmd.Env = env
+	}
+
+	// Add shared data as environment variables
+	if work.Execution.ExecutionOptions != nil && work.Execution.ExecutionOptions.SharedData != nil {
+		env := cmd.Env
+		if env == nil {
+			env = os.Environ()
+		}
+		for key, value := range work.Execution.ExecutionOptions.SharedData {
+			env = append(env, fmt.Sprintf("YAO_JOB_SHARED_%s=%v", key, value))
+		}
+		cmd.Env = env
+	}
+
+	// Execute command with context cancellation support
+	output, err := cmd.CombinedOutput()
+	if err != nil {
+		// Check if it was cancelled
+		if ctx.Err() != nil {
+			work.Execution.Warn("Command cancelled: %s", ctx.Err().Error())
+			work.Execution.Status = "cancelled"
+		} else {
+			work.Execution.Error("Command failed: %s, output: %s", err.Error(), string(output))
+			work.Execution.Status = "failed"
+
+			// Store error output
+			if len(output) > 0 {
+				errorInfo := map[string]interface{}{
+					"error":  err.Error(),
+					"output": string(output),
+				}
+				if errorBytes, jsonErr := jsoniter.Marshal(errorInfo); jsonErr == nil {
+					work.Execution.ErrorInfo = (*json.RawMessage)(&errorBytes)
+				}
+			}
+		}
+
+		// Save execution status
+		if saveErr := SaveExecution(work.Execution); saveErr != nil {
+			work.Execution.Error("Failed to save execution error: %s", saveErr.Error())
+		}
+
+		if ctx.Err() != nil {
+			return ctx.Err()
+		}
+		return fmt.Errorf("command execution failed: %v", err)
+	}
+
+	work.Execution.Info("Command completed successfully, output: %s", string(output))
+
+	// Update execution with success result
+	work.Execution.Status = "completed"
+	work.Execution.Progress = 100
+	if len(output) > 0 {
+		result := map[string]interface{}{
+			"output": string(output),
+		}
+		if resultBytes, err := jsoniter.Marshal(result); err == nil {
+			work.Execution.Result = (*json.RawMessage)(&resultBytes)
+		}
+	}
+
+	if saveErr := SaveExecution(work.Execution); saveErr != nil {
+		work.Execution.Error("Failed to save execution result: %s", saveErr.Error())
+		return fmt.Errorf("failed to save execution result: %w", saveErr)
+	}
+
+	return nil
+}
+
+// UpdateExecutionProgress updates execution progress from external callback
+// This function can be called by system commands via HTTP API to report progress
+func UpdateExecutionProgress(executionID string, progressData map[string]interface{}) error {
+	// Load execution from database
+	execution, err := GetExecution(executionID, model.QueryParam{})
+	if err != nil {
+		return fmt.Errorf("failed to load execution: %w", err)
+	}
+
+	if execution == nil {
+		return fmt.Errorf("execution not found: %s", executionID)
+	}
+
+	// Extract progress and message from callback data
+	if progressVal, exists := progressData["progress"]; exists {
+		if progress, ok := progressVal.(float64); ok {
+			execution.Progress = int(progress)
+		} else if progress, ok := progressVal.(int); ok {
+			execution.Progress = progress
+		}
+	}
+
+	if messageVal, exists := progressData["message"]; exists {
+		if message, ok := messageVal.(string); ok {
+			execution.Info("Progress update: %s (%.1f%%)", message, float64(execution.Progress))
+		}
+	}
+
+	// Save updated execution to database
+	if saveErr := SaveExecution(execution); saveErr != nil {
+		return fmt.Errorf("failed to save progress update: %w", saveErr)
+	}
+
+	return nil
+}
+
+// extractProgressData extracts progress and message from callback data
+func extractProgressData(data map[string]interface{}) (int, string) {
+	var progressInt int = -1 // Default to -1 to indicate no progress value
+	var message string
+
+	// Extract progress value with type assertion
+	if progressVal, exists := data["progress"]; exists {
+		switch v := progressVal.(type) {
+		case float64:
+			progressInt = int(v)
+		case int:
+			progressInt = v
+		case int32:
+			progressInt = int(v)
+		case int64:
+			progressInt = int(v)
+		}
+	}
+
+	// Extract message with type assertion
+	if messageVal, exists := data["message"]; exists {
+		if msg, ok := messageVal.(string); ok {
+			message = msg
+		}
+	}
+
+	return progressInt, message
+}

job/job.go

@@ -1,11 +1,15 @@
 package job
 
 import (
+	"context"
 	"fmt"
+	"sort"
+	"strings"
 	"time"
 
 	jsoniter "github.com/json-iterator/go"
 	"github.com/yaoapp/gou/model"
+	"github.com/yaoapp/kun/log"
 )
 
 // Once create a new job
@@ -42,17 +46,39 @@ func Daemon(mode ModeType, data map[string]interface{}) (*Job, error) {
 	return makeJob(raw)
 }
 
-// SetWorkerManager sets a custom worker manager for this job (for testing)
-func (j *Job) SetWorkerManager(wm *WorkerManager) {
-	j.workerManager = wm
-}
-
 // Start start the job
 func (j *Job) Start() error {
-	// Get handler from registry (thread-safe)
+	// Get executions for this job
-	handler, exists := getHandler(j.JobID)
+	executions, err := j.GetExecutions()
-	if !exists {
+	if err != nil {
-		return fmt.Errorf("no handler registered for job %s", j.JobID)
+		return fmt.Errorf("failed to get executions: %w", err)
+	}
+
+	if len(executions) == 0 {
+		return fmt.Errorf("no executions found for job %s", j.JobID)
+	}
+
+	// Sort executions by priority (higher priority first)
+	sort.Slice(executions, func(i, j int) bool {
+		priorityI := 0
+		if executions[i].ExecutionOptions != nil {
+			priorityI = executions[i].ExecutionOptions.Priority
+		}
+		priorityJ := 0
+		if executions[j].ExecutionOptions != nil {
+			priorityJ = executions[j].ExecutionOptions.Priority
+		}
+		return priorityI > priorityJ
+	})
+
+	// Initialize job context for cancellation
+	if j.ctx == nil {
+		j.ctx, j.cancel = context.WithCancel(context.Background())
+	}
+
+	// Initialize execution contexts map
+	if j.executionContexts == nil {
+		j.executionContexts = make(map[string]context.CancelFunc)
 	}
 
 	// Update job status to ready
@@ -61,32 +87,62 @@ func (j *Job) Start() error {
 		return fmt.Errorf("failed to update job status: %w", err)
 	}
 
-	// Submit to worker manager (use custom one if set, otherwise global)
+	// Get global worker manager (should be already started)
-	var wm *WorkerManager
+	wm := GetWorkerManager()
-	if j.workerManager != nil {
+
-		wm = j.workerManager
+	// Submit executions and ensure all are added successfully
-	} else {
+	var submitErrors []string
-		wm = GetWorkerManager()
+	for _, execution := range executions {
-		// Start worker manager if not already started
+		// Create execution-specific context derived from job context
-		if wm.GetActiveWorkers() == 0 {
+		execCtx, execCancel := context.WithCancel(j.ctx)
-			wm.Start()
+
+		// Store execution cancel function
+		j.executionMutex.Lock()
+		j.executionContexts[execution.ExecutionID] = execCancel
+		j.executionMutex.Unlock()
+
+		// Submit execution (non-blocking)
+		if err := wm.SubmitJob(execCtx, j, execution); err != nil {
+			// Clean up on error
+			execCancel()
+			j.executionMutex.Lock()
+			delete(j.executionContexts, execution.ExecutionID)
+			j.executionMutex.Unlock()
+
+			submitErrors = append(submitErrors, fmt.Sprintf("execution %s: %v", execution.ExecutionID, err))
+			log.Error("Failed to submit execution %s: %v", execution.ExecutionID, err)
 		}
 	}
 
-	return wm.SubmitJob(j, handler)
+	// Return error if any submissions failed
+	if len(submitErrors) > 0 {
+		return fmt.Errorf("failed to submit some executions: %s", strings.Join(submitErrors, "; "))
+	}
+
+	return nil
 }
 
-// Cancel cancel the job
+// Stop stops the job and cancels all running executions
-func (j *Job) Cancel() error {
+func (j *Job) Stop() error {
 	// Update job status
 	j.Status = "disabled"
 	if err := SaveJob(j); err != nil {
 		return fmt.Errorf("failed to update job status: %w", err)
 	}
 
-	// If there's a current execution, mark it as cancelled
+	// Cancel all running executions using job context
-	if j.CurrentExecutionID != nil {
+	if j.cancel != nil {
-		execution, err := GetExecution(*j.CurrentExecutionID, model.QueryParam{})
+		j.cancel()
+		log.Info("Job %s cancelled, all executions will be stopped", j.JobID)
+	}
+
+	// Cancel individual execution contexts and clean up
+	j.executionMutex.Lock()
+	for executionID, cancelFunc := range j.executionContexts {
+		cancelFunc()
+
+		// Update execution status in database
+		execution, err := GetExecution(executionID, model.QueryParam{})
 		if err == nil && (execution.Status == "queued" || execution.Status == "running") {
 			execution.Status = "cancelled"
 			execution.EndedAt = &time.Time{}
@@ -98,17 +154,45 @@ func (j *Job) Cancel() error {
 				JobID:       j.JobID,
 				Level:       "info",
 				Message:     "Job execution cancelled by user",
-				ExecutionID: j.CurrentExecutionID,
+				ExecutionID: &executionID,
 				Timestamp:   time.Now(),
 				Sequence:    0,
 			}
 			SaveLog(logEntry)
 		}
 	}
+	// Clear execution contexts
+	j.executionContexts = make(map[string]context.CancelFunc)
+	j.executionMutex.Unlock()
 
 	return nil
 }
 
+// Destroy destroys the job and cleans up all resources
+func (j *Job) Destroy() error {
+	// Stop the job first
+	if err := j.Stop(); err != nil {
+		log.Warn("Failed to stop job during destroy: %v", err)
+	}
+
+	// Handlers are now stored with executions, no global registry to clean
+
+	// Update job status to deleted
+	j.Status = "deleted"
+	if err := SaveJob(j); err != nil {
+		log.Warn("Failed to update job status to deleted: %v", err)
+	}
+
+	// Clear job references
+	if j.cancel != nil {
+		j.cancel()
+		j.cancel = nil
+	}
+
+	log.Info("Job %s destroyed successfully", j.JobID)
+	return nil
+}
+
 // SetData set the data of the job
 func (j *Job) SetData(data map[string]interface{}) *Job {
 	return j
@@ -173,5 +257,38 @@ func makeJob(data []byte) (*Job, error) {
 	if err != nil {
 		return nil, err
 	}
+
+	// Set default values if not provided
+	if job.CategoryID == "" {
+		job.CategoryID = "default"
+	}
+	if job.Status == "" {
+		job.Status = "draft"
+	}
+	if job.MaxWorkerNums == 0 {
+		job.MaxWorkerNums = 1 // Default to 1 worker
+	}
+	if job.Priority == 0 {
+		job.Priority = 1 // Default job priority
+	}
+	if job.CreatedBy == "" {
+		job.CreatedBy = "system"
+	}
+
 	return &job, nil
 }
+
+// RestoreJobsFromDatabase restores jobs from database on system startup
+func RestoreJobsFromDatabase() ([]*Job, error) {
+	// Get all active jobs from database
+	activeJobs, err := GetActiveJobs()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get active jobs: %w", err)
+	}
+
+	// No need to restore handlers since we only use Yao processes and commands
+	// Both are fully serializable and self-contained
+
+	log.Info("Restored %d jobs from database", len(activeJobs))
+	return activeJobs, nil
+}

job/job_test.go

@@ -1,68 +1,178 @@
 package job_test
 
 import (
-	"context"
 	"fmt"
 	"testing"
 	"time"
 
 	"github.com/yaoapp/gou/model"
+	"github.com/yaoapp/gou/process"
 	"github.com/yaoapp/yao/config"
 	"github.com/yaoapp/yao/job"
 	"github.com/yaoapp/yao/test"
 )
 
+// registerTestProcesses registers test processes for job testing
+func registerTestProcesses() {
+	// Register test.job.echo process
+	process.Register("test.job.echo", func(process *process.Process) interface{} {
+		args := process.Args
+		if len(args) > 0 {
+			message := args[0]
+
+			// Simulate progress updates
+			if process.Callback != nil {
+				// Report 25% progress
+				process.Callback(process, map[string]interface{}{
+					"type":     "progress",
+					"progress": 25,
+					"message":  "Starting echo process",
+				})
+
+				// Report 50% progress
+				process.Callback(process, map[string]interface{}{
+					"type":     "progress",
+					"progress": 50,
+					"message":  "Processing message",
+				})
+
+				// Report 75% progress
+				process.Callback(process, map[string]interface{}{
+					"type":     "progress",
+					"progress": 75,
+					"message":  "Finalizing echo",
+				})
+
+				// Report 100% progress
+				process.Callback(process, map[string]interface{}{
+					"type":     "progress",
+					"progress": 100,
+					"message":  "Echo completed",
+				})
+			}
+
+			return map[string]interface{}{
+				"message": message,
+				"echo":    "Echo: " + message.(string),
+				"status":  "success",
+			}
+		}
+
+		return map[string]interface{}{
+			"message": "No message provided",
+			"status":  "error",
+		}
+	})
+
+	// Register test.job.cron process
+	process.Register("test.job.cron", func(process *process.Process) interface{} {
+		args := process.Args
+		message := "Cron job executed"
+		if len(args) > 0 {
+			message = args[0].(string)
+		}
+
+		return map[string]interface{}{
+			"message":   message,
+			"timestamp": time.Now().Unix(),
+			"status":    "success",
+		}
+	})
+
+	// Register test.job.daemon process
+	process.Register("test.job.daemon", func(process *process.Process) interface{} {
+		args := process.Args
+		message := "Daemon process executed"
+		if len(args) > 0 {
+			message = args[0].(string)
+		}
+
+		return map[string]interface{}{
+			"message": message,
+			"status":  "success",
+			"daemon":  true,
+		}
+	})
+
+	// Register test.job.database process
+	process.Register("test.job.database", func(process *process.Process) interface{} {
+		args := process.Args
+		message := "Database operation executed"
+		if len(args) > 0 {
+			message = args[0].(string)
+		}
+
+		return map[string]interface{}{
+			"message":   message,
+			"operation": "test",
+			"status":    "success",
+		}
+	})
+
+	// Register test.job.execution process with enhanced features
+	process.Register("test.job.execution", func(process *process.Process) interface{} {
+		args := process.Args
+		message := "Execution test"
+		if len(args) > 0 {
+			message = args[0].(string)
+		}
+
+		// Simulate progress updates with callback
+		if process.Callback != nil {
+			// Report progress incrementally
+			for i := 10; i <= 100; i += 10 {
+				process.Callback(process, map[string]interface{}{
+					"type":     "progress",
+					"progress": i,
+					"message":  fmt.Sprintf("Processing step %d/10", i/10),
+				})
+				time.Sleep(10 * time.Millisecond) // Small delay to simulate work
+			}
+		}
+
+		return map[string]interface{}{
+			"message":   message,
+			"progress":  100,
+			"status":    "success",
+			"test_data": "execution completed",
+		}
+	})
+}
+
 // TestOnce test once job
 func TestOnceGoroutine(t *testing.T) {
 	// Setup
 	test.Prepare(&testing.T{}, config.Conf)
 	defer test.Clean()
 
+	// Register test processes
+	registerTestProcesses()
+
 	testJob, err := job.Once(job.GOROUTINE, map[string]interface{}{})
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	// Channel to signal completion
+	// Use a test Yao process (this would need to be defined in your Yao app)
-	done := make(chan bool, 1)
+	err = testJob.Add(&job.ExecutionOptions{
-
+		Priority: 1,
-	// Handler that signals completion
+		SharedData: map[string]interface{}{
-	handler := func(ctx context.Context, execution *job.Execution) error {
+			"test_data": "Hello from test",
-		execution.SetProgress(50, "Progress 50%%")
+		},
-		execution.Info("Progress 50%%")
+	}, "test.job.echo", "Hello from test")
-		time.Sleep(100 * time.Millisecond)
-		execution.SetProgress(100, "Progress 100%%")
-		execution.Info("Progress 100%%")
-		done <- true
-		return nil
-	}
-
-	err = testJob.Add(1, handler)
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	// Set up worker manager for test
-	wm := job.NewWorkerManagerForTest(2)
-	wm.Start()
-	defer wm.Stop()
-	testJob.SetWorkerManager(wm)
-
 	err = testJob.Start()
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	// Wait for job completion or timeout
+	// Give some time for execution
-	select {
+	time.Sleep(2 * time.Second)
-	case <-done:
-		t.Log("Job completed successfully")
-	case <-time.After(10 * time.Second):
-		t.Error("Job execution timeout")
-	}
 
-	// Give some extra time for cleanup
+	t.Log("Job started successfully")
-	time.Sleep(500 * time.Millisecond)
 }
 
 func TestOnceProcess(t *testing.T) {
@@ -70,51 +180,34 @@ func TestOnceProcess(t *testing.T) {
 	test.Prepare(&testing.T{}, config.Conf)
 	defer test.Clean()
 
+	// Register test processes
+	registerTestProcesses()
+
 	testJob, err := job.Once(job.PROCESS, map[string]interface{}{})
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	// Channel to signal completion
+	// Use a test Yao process
-	done := make(chan bool, 1)
+	err = testJob.Add(&job.ExecutionOptions{
-
+		Priority: 1,
-	// Handler that signals completion
+		SharedData: map[string]interface{}{
-	handler := func(ctx context.Context, execution *job.Execution) error {
+			"test_data": "Hello from process test",
-		execution.SetProgress(50, "Progress 50%%")
+		},
-		execution.Info("Progress 50%%")
+	}, "test.job.echo", "Hello from process test")
-		time.Sleep(100 * time.Millisecond)
-		execution.SetProgress(100, "Progress 100%%")
-		execution.Info("Progress 100%%")
-		done <- true
-		return nil
-	}
-
-	err = testJob.Add(1, handler)
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	// Set up worker manager for test
-	wm := job.NewWorkerManagerForTest(2)
-	wm.Start()
-	defer wm.Stop()
-	testJob.SetWorkerManager(wm)
-
 	err = testJob.Start()
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	// Wait for job completion or timeout
+	// Give some time for execution
-	select {
+	time.Sleep(2 * time.Second)
-	case <-done:
-		t.Log("Job completed successfully")
-	case <-time.After(10 * time.Second):
-		t.Error("Job execution timeout")
-	}
 
-	// Give some extra time for cleanup
+	t.Log("Process job started successfully")
-	time.Sleep(500 * time.Millisecond)
 }
 
 func TestCronGoroutine(t *testing.T) {
@@ -122,13 +215,21 @@ func TestCronGoroutine(t *testing.T) {
 	test.Prepare(&testing.T{}, config.Conf)
 	defer test.Clean()
 
+	// Register test processes
+	registerTestProcesses()
+
 	testJob, err := job.Cron(job.GOROUTINE, map[string]interface{}{}, "0 0 * * *")
 	if err != nil {
 		t.Fatal(err)
 	}
 
 	// For cron jobs, we just test creation, not execution
-	err = testJob.Add(1, HandlerTest)
+	err = testJob.Add(&job.ExecutionOptions{
+		Priority: 1,
+		SharedData: map[string]interface{}{
+			"cron_context": "scheduled execution",
+		},
+	}, "test.job.cron", "Cron test execution")
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -145,13 +246,21 @@ func TestCronProcess(t *testing.T) {
 	test.Prepare(&testing.T{}, config.Conf)
 	defer test.Clean()
 
+	// Register test processes
+	registerTestProcesses()
+
 	testJob, err := job.Cron(job.PROCESS, map[string]interface{}{}, "0 0 * * *")
 	if err != nil {
 		t.Fatal(err)
 	}
 
 	// For cron jobs, we just test creation, not execution
-	err = testJob.Add(1, HandlerTest)
+	err = testJob.Add(&job.ExecutionOptions{
+		Priority: 1,
+		SharedData: map[string]interface{}{
+			"cron_context": "scheduled process execution",
+		},
+	}, "test.job.cron", "Cron process test execution")
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -163,18 +272,26 @@ func TestCronProcess(t *testing.T) {
 	}
 }
 
-// TestDaemonGoroutine tests daemon job with goroutine mode using Ticker handler
+// TestDaemonGoroutine tests daemon job with goroutine mode
 func TestDaemonGoroutine(t *testing.T) {
 	test.Prepare(&testing.T{}, config.Conf)
 	defer test.Clean()
 
+	// Register test processes
+	registerTestProcesses()
+
 	testJob, err := job.Daemon(job.GOROUTINE, map[string]interface{}{})
 	if err != nil {
 		t.Fatal(err)
 	}
 
 	// For daemon jobs, we just test creation, not long-running execution
-	err = testJob.Add(1, DaemonHandlerFastTest)
+	err = testJob.Add(&job.ExecutionOptions{
+		Priority: 1,
+		SharedData: map[string]interface{}{
+			"daemon_context": "background service",
+		},
+	}, "test.job.daemon", "Daemon test execution")
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -186,18 +303,26 @@ func TestDaemonGoroutine(t *testing.T) {
 	}
 }
 
-// TestDaemonProcess tests daemon job with process mode using Ticker handler
+// TestDaemonProcess tests daemon job with process mode
 func TestDaemonProcess(t *testing.T) {
 	test.Prepare(&testing.T{}, config.Conf)
 	defer test.Clean()
 
+	// Register test processes
+	registerTestProcesses()
+
 	testJob, err := job.Daemon(job.PROCESS, map[string]interface{}{})
 	if err != nil {
 		t.Fatal(err)
 	}
 
 	// For daemon jobs, we just test creation, not long-running execution
-	err = testJob.Add(1, DaemonHandlerFastTest)
+	err = testJob.Add(&job.ExecutionOptions{
+		Priority: 1,
+		SharedData: map[string]interface{}{
+			"daemon_context": "background process service",
+		},
+	}, "test.job.daemon", "Daemon process test execution")
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -209,132 +334,42 @@ func TestDaemonProcess(t *testing.T) {
 	}
 }
 
-// TestDaemonFastGoroutine tests fast daemon job with goroutine mode for quick testing
+// TestCommand test command execution
-func TestDaemonFastGoroutine(t *testing.T) {
+func TestCommand(t *testing.T) {
 	test.Prepare(&testing.T{}, config.Conf)
 	defer test.Clean()
 
-	testJob, err := job.Daemon(job.GOROUTINE, map[string]interface{}{})
+	// Register test processes
+	registerTestProcesses()
+
+	testJob, err := job.Once(job.GOROUTINE, map[string]interface{}{
+		"name":        "Test Command Job",
+		"description": "Job for testing command execution",
+	})
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	// For daemon jobs, we just test creation, not execution
+	// Test system command
-	err = testJob.Add(1, DaemonHandlerFastTest)
+	err = testJob.AddCommand(&job.ExecutionOptions{
+		Priority: 1,
+		SharedData: map[string]interface{}{
+			"command_context": "test execution",
+		},
+	}, "echo", []string{"Hello from command test"}, nil)
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	// Just verify the job was created properly
+	err = testJob.Start()
-	if testJob.ScheduleType != string(job.ScheduleTypeDaemon) {
-		t.Errorf("Expected schedule type daemon, got %s", testJob.ScheduleType)
-	}
-}
-
-// TestDaemonFastProcess tests fast daemon job with process mode for quick testing
-func TestDaemonFastProcess(t *testing.T) {
-	test.Prepare(&testing.T{}, config.Conf)
-	defer test.Clean()
-
-	testJob, err := job.Daemon(job.PROCESS, map[string]interface{}{})
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	// For daemon jobs, we just test creation, not execution
+	// Give some time for execution
-	err = testJob.Add(1, DaemonHandlerFastTest)
+	time.Sleep(2 * time.Second)
-	if err != nil {
-		t.Fatal(err)
-	}
 
-	// Just verify the job was created properly
+	t.Log("Command job started successfully")
-	if testJob.ScheduleType != string(job.ScheduleTypeDaemon) {
-		t.Errorf("Expected schedule type daemon, got %s", testJob.ScheduleType)
-	}
-}
-
-func HandlerTest(ctx context.Context, execution *job.Execution) error {
-	execution.SetProgress(50, "Progress 50%%")
-	execution.Info("Progress 50%%")
-	time.Sleep(100 * time.Millisecond)
-	execution.SetProgress(100, "Progress 100%%")
-	execution.Info("Progress 100%%")
-	time.Sleep(200 * time.Millisecond)
-	execution.SetProgress(100, "Progress 100%%")
-	execution.Info("Progress 100%%")
-	return nil
-}
-
-func DaemonHandlerTest(ctx context.Context, execution *job.Execution) error {
-	// Build a daemon handler using Ticker that executes tasks every 5 seconds continuously
-	ticker := time.NewTicker(5 * time.Second)
-	defer ticker.Stop()
-
-	counter := 0
-
-	execution.Info("Daemon handler started, running continuously...")
-	execution.SetProgress(0, "Daemon initialized and ready")
-
-	for {
-		select {
-		case <-ctx.Done():
-			// Context cancelled, graceful shutdown
-			execution.Info("Daemon handler received cancellation signal after %d iterations, exiting...", counter)
-			execution.SetProgress(100, fmt.Sprintf("Daemon stopped gracefully after %d iterations", counter))
-			return ctx.Err()
-		case <-ticker.C:
-			counter++
-
-			// Daemon doesn't need specific completion progress, show running status instead
-			execution.SetProgress(50, fmt.Sprintf("Running - completed %d iterations", counter))
-			execution.Info("Daemon tick %d: Processing periodic task...", counter)
-
-			// Simulate periodic tasks execution
-			// e.g.: cleanup temp files, health checks, data synchronization, etc.
-			time.Sleep(500 * time.Millisecond) // Simulate task execution time
-
-			execution.Debug("Daemon iteration %d completed successfully", counter)
-
-			// Output statistics every 10 iterations
-			if counter%10 == 0 {
-				execution.Info("Daemon health check: %d iterations completed, still running...", counter)
-			}
-		}
-	}
-}
-
-// DaemonHandlerFastTest fast testing version of daemon handler for testing (executes every 500ms)
-func DaemonHandlerFastTest(ctx context.Context, execution *job.Execution) error {
-	// Use shorter interval for testing
-	ticker := time.NewTicker(500 * time.Millisecond)
-	defer ticker.Stop()
-
-	counter := 0
-
-	execution.Info("Fast daemon handler started for testing, running continuously...")
-	execution.SetProgress(0, "Fast daemon initialized")
-
-	for {
-		select {
-		case <-ctx.Done():
-			execution.Info("Fast daemon handler stopped after %d iterations", counter)
-			execution.SetProgress(100, fmt.Sprintf("Fast daemon stopped after %d iterations", counter))
-			return ctx.Err()
-		case <-ticker.C:
-			counter++
-
-			execution.SetProgress(50, fmt.Sprintf("Fast daemon: %d iterations", counter))
-			execution.Debug("Fast daemon tick %d: Quick task execution", counter)
-
-			// Quick task simulation
-			time.Sleep(50 * time.Millisecond)
-
-			// Output info every 5 iterations (due to higher frequency)
-			if counter%5 == 0 {
-				execution.Info("Fast daemon: %d iterations completed", counter)
-			}
-		}
-	}
 }
 
 // TestDatabase test database operations
@@ -342,6 +377,9 @@ func TestDatabase(t *testing.T) {
 	test.Prepare(t, config.Conf)
 	defer test.Clean()
 
+	// Register test processes
+	registerTestProcesses()
+
 	// Test category creation
 	category, err := job.GetOrCreateCategory("test-category", "Test category for unit tests")
 	if err != nil {
@@ -362,7 +400,18 @@ func TestDatabase(t *testing.T) {
 	}
 
 	testJob.SetCategory(category.CategoryID)
-	testJob.Add(1, HandlerTest)
+	testJob.Add(&job.ExecutionOptions{
+		Priority: 1,
+		SharedData: map[string]interface{}{
+			"database_context": "test operation",
+		},
+	}, "test.job.database", "Database test execution")
+
+	// Save the job to database before retrieving it
+	err = job.SaveJob(testJob)
+	if err != nil {
+		t.Fatalf("Failed to save job: %v", err)
+	}
 
 	// Test job retrieval
 	retrievedJob, err := job.GetJob(testJob.JobID)
@@ -398,71 +447,14 @@ func TestDatabase(t *testing.T) {
 	}
 }
 
-// TestWorkerManager test worker management
-func TestWorkerManager(t *testing.T) {
-	test.Prepare(t, config.Conf)
-	defer test.Clean()
-
-	// Get worker manager
-	wm := job.NewWorkerManagerForTest(2)
-	if wm == nil {
-		t.Fatal("Failed to get worker manager")
-	}
-
-	// Start worker manager
-	wm.Start()
-	defer wm.Stop()
-
-	// Check active workers
-	activeWorkers := wm.GetActiveWorkers()
-	if activeWorkers == 0 {
-		t.Error("Expected active workers after starting worker manager")
-	}
-
-	// Create and submit a job
-	testJob, err := job.Once(job.GOROUTINE, map[string]interface{}{
-		"name":        "Test Worker Job",
-		"description": "Job for testing worker management",
-	})
-	if err != nil {
-		t.Fatalf("Failed to create job: %v", err)
-	}
-
-	err = testJob.Add(1, HandlerTest)
-	if err != nil {
-		t.Fatalf("Failed to add handler: %v", err)
-	}
-
-	// Start the job
-	err = testJob.Start()
-	if err != nil {
-		t.Fatalf("Failed to start job: %v", err)
-	}
-
-	// Wait for job to complete
-	time.Sleep(1 * time.Second)
-
-	// Check executions
-	executions, err := testJob.GetExecutions()
-	if err != nil {
-		t.Fatalf("Failed to get executions: %v", err)
-	}
-
-	if len(executions) == 0 {
-		t.Error("Expected at least one execution")
-	}
-
-	// Check execution status
-	if executions[0].Status != "completed" && executions[0].Status != "running" {
-		t.Errorf("Expected execution status 'completed' or 'running', got '%s'", executions[0].Status)
-	}
-}
-
 // TestJobExecution test job execution with logging and progress
 func TestJobExecution(t *testing.T) {
 	test.Prepare(t, config.Conf)
 	defer test.Clean()
 
+	// Register test processes
+	registerTestProcesses()
+
 	// Create a job with enhanced handler
 	testJob, err := job.Once(job.GOROUTINE, map[string]interface{}{
 		"name":        "Test Execution Job",
@@ -472,56 +464,36 @@ func TestJobExecution(t *testing.T) {
 		t.Fatalf("Failed to create job: %v", err)
 	}
 
-	// Channel to signal completion
+	// Save the job to database first so it has a valid ID
-	done := make(chan bool, 1)
+	err = job.SaveJob(testJob)
-
+	if err != nil {
-	// Enhanced handler for testing
+		t.Fatalf("Failed to save job: %v", err)
-	enhancedHandler := func(ctx context.Context, execution *job.Execution) error {
-		execution.Info("Starting enhanced test execution")
-		execution.SetProgress(10, "Initialization complete")
-
-		time.Sleep(50 * time.Millisecond)
-
-		execution.Debug("Debug message test")
-		execution.SetProgress(50, "Halfway complete")
-
-		time.Sleep(50 * time.Millisecond)
-
-		execution.Warn("Warning message test")
-		execution.SetProgress(80, "Almost done")
-
-		time.Sleep(50 * time.Millisecond)
-
-		execution.Info("Execution completed successfully")
-		execution.SetProgress(100, "Complete")
-
-		done <- true
-		return nil
 	}
 
-	err = testJob.Add(1, enhancedHandler)
+	// Use a test Yao process for execution testing with chained options
+	err = testJob.Add(
+		job.NewExecutionOptions().
+			WithPriority(1).
+			AddSharedData("execution_context", "enhanced test").
+			AddSharedData("user_id", "test_user_123").
+			AddSharedData("session", map[string]interface{}{
+				"token":   "test_token",
+				"expires": "2024-12-31",
+			}),
+		"test.job.execution", "Enhanced execution test")
 	if err != nil {
 		t.Fatalf("Failed to add handler: %v", err)
 	}
 
-	// Start worker manager
-	wm := job.NewWorkerManagerForTest(2)
-	wm.Start()
-	defer wm.Stop()
-
 	// Start the job
 	err = testJob.Start()
 	if err != nil {
 		t.Fatalf("Failed to start job: %v", err)
 	}
 
-	// Wait for completion or timeout
+	// Give some time for execution
-	select {
+	time.Sleep(2 * time.Second)
-	case <-done:
+	t.Log("Job execution started")
-		t.Log("Job execution completed")
-	case <-time.After(10 * time.Second):
-		t.Error("Job execution timeout")
-	}
 
 	// Give some extra time for database operations to complete
 	time.Sleep(200 * time.Millisecond)
@@ -537,10 +509,25 @@ func TestJobExecution(t *testing.T) {
 	}
 
 	execution := executions[0]
+	t.Logf("Initial execution progress: %d", execution.Progress)
+
+	// Get fresh execution data from database to check final progress
+	freshExecution, err := job.GetExecution(execution.ExecutionID, model.QueryParam{})
+	if err != nil {
+		t.Fatalf("Failed to get fresh execution: %v", err)
+	}
+
+	t.Logf("Fresh execution progress: %d, status: %s", freshExecution.Progress, freshExecution.Status)
+	if freshExecution.ErrorInfo != nil && len(*freshExecution.ErrorInfo) > 0 {
+		t.Logf("Execution error: %s", string(*freshExecution.ErrorInfo))
+	}
+	if freshExecution.Result != nil && len(*freshExecution.Result) > 0 {
+		t.Logf("Execution result: %s", string(*freshExecution.Result))
+	}
 
 	// Check final progress (may take time to update)
-	if execution.Progress < 50 {
+	if freshExecution.Progress < 50 {
-		t.Errorf("Expected progress at least 50, got %d", execution.Progress)
+		t.Errorf("Expected progress at least 50, got %d", freshExecution.Progress)
 	}
 
 	// Check logs

job/process.go

@@ -1,4 +1,247 @@
 package job
 
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+	"os/exec"
+
+	jsoniter "github.com/json-iterator/go"
+	"github.com/yaoapp/yao/config"
+)
+
 // Process the process mode
 type Process struct{}
+
+// ExecuteYaoProcess executes a Yao process using independent process mode (yao run command)
+func (p *Process) ExecuteYaoProcess(ctx context.Context, work *WorkRequest, progress *Progress) error {
+	execConfig := work.Execution.ExecutionConfig
+
+	work.Execution.Info("Executing Yao process: %s (process mode)", execConfig.ProcessName)
+
+	// Prepare yao run command arguments
+	args := []string{"run", execConfig.ProcessName}
+
+	// Convert and add process arguments using the proper conversion function
+	convertedArgs := convertArgsForYaoRun(execConfig.ProcessArgs)
+	args = append(args, convertedArgs...)
+
+	// Create command with context for cancellation support
+	cmd := exec.CommandContext(ctx, "yao", args...)
+
+	// Set working directory to Yao application root
+	if config.Conf.Root != "" {
+		cmd.Dir = config.Conf.Root
+	} else {
+		// Fallback to current directory if config is not available
+		cmd.Dir, _ = os.Getwd()
+	}
+
+	// Set environment variables
+	env := os.Environ()
+	env = append(env,
+		fmt.Sprintf("YAO_JOB_ID=%s", work.Job.JobID),
+		fmt.Sprintf("YAO_EXECUTION_ID=%s", work.Execution.ExecutionID),
+	)
+
+	// Add shared data as environment variables
+	if work.Execution.ExecutionOptions != nil && work.Execution.ExecutionOptions.SharedData != nil {
+		for key, value := range work.Execution.ExecutionOptions.SharedData {
+			if valueBytes, err := jsoniter.Marshal(value); err == nil {
+				env = append(env, fmt.Sprintf("YAO_JOB_SHARED_%s=%s", key, string(valueBytes)))
+			} else {
+				env = append(env, fmt.Sprintf("YAO_JOB_SHARED_%s=%v", key, value))
+			}
+		}
+	}
+	cmd.Env = env
+
+	// Execute command
+	output, err := cmd.CombinedOutput()
+	if err != nil {
+		// Check if it was cancelled
+		if ctx.Err() != nil {
+			work.Execution.Warn("Yao process cancelled: %s", ctx.Err().Error())
+			work.Execution.Status = "cancelled"
+		} else {
+			work.Execution.Error("Yao process failed: %s, output: %s", err.Error(), string(output))
+			work.Execution.Status = "failed"
+
+			// Store error output
+			if len(output) > 0 {
+				errorInfo := map[string]interface{}{
+					"error":  err.Error(),
+					"output": string(output),
+				}
+				if errorBytes, jsonErr := jsoniter.Marshal(errorInfo); jsonErr == nil {
+					work.Execution.ErrorInfo = (*json.RawMessage)(&errorBytes)
+				}
+			}
+		}
+
+		// Save execution status
+		if saveErr := SaveExecution(work.Execution); saveErr != nil {
+			work.Execution.Error("Failed to save execution error: %s", saveErr.Error())
+		}
+
+		if ctx.Err() != nil {
+			return ctx.Err()
+		}
+		return fmt.Errorf("yao process execution failed: %v", err)
+	}
+
+	work.Execution.Info("Yao process completed successfully, output: %s", string(output))
+
+	// Update execution with success result
+	work.Execution.Status = "completed"
+	work.Execution.Progress = 100
+	if len(output) > 0 {
+		result := map[string]interface{}{
+			"output": string(output),
+		}
+		if resultBytes, err := jsoniter.Marshal(result); err == nil {
+			work.Execution.Result = (*json.RawMessage)(&resultBytes)
+		}
+	}
+
+	if saveErr := SaveExecution(work.Execution); saveErr != nil {
+		work.Execution.Error("Failed to save execution result: %s", saveErr.Error())
+		return fmt.Errorf("failed to save execution result: %w", saveErr)
+	}
+
+	return nil
+}
+
+// ExecuteSystemCommand executes a system command using independent process mode
+func (p *Process) ExecuteSystemCommand(ctx context.Context, work *WorkRequest, progress *Progress) error {
+	execConfig := work.Execution.ExecutionConfig
+
+	work.Execution.Info("Executing command: %s (process mode)", execConfig.Command)
+
+	// Create command with context for cancellation support
+	cmd := exec.CommandContext(ctx, execConfig.Command, execConfig.CommandArgs...)
+
+	// Set working directory to Yao application root directory
+	if config.Conf.Root != "" {
+		cmd.Dir = config.Conf.Root
+	} else {
+		// Fallback to current directory
+		cmd.Dir, _ = os.Getwd()
+	}
+
+	// Set environment variables
+	env := os.Environ()
+	if len(execConfig.Environment) > 0 {
+		for key, value := range execConfig.Environment {
+			env = append(env, fmt.Sprintf("%s=%s", key, value))
+		}
+	}
+
+	// Add job context
+	env = append(env,
+		fmt.Sprintf("YAO_JOB_ID=%s", work.Job.JobID),
+		fmt.Sprintf("YAO_EXECUTION_ID=%s", work.Execution.ExecutionID),
+	)
+
+	// Add shared data as environment variables
+	if work.Execution.ExecutionOptions != nil && work.Execution.ExecutionOptions.SharedData != nil {
+		for key, value := range work.Execution.ExecutionOptions.SharedData {
+			if valueBytes, err := jsoniter.Marshal(value); err == nil {
+				env = append(env, fmt.Sprintf("YAO_JOB_SHARED_%s=%s", key, string(valueBytes)))
+			} else {
+				env = append(env, fmt.Sprintf("YAO_JOB_SHARED_%s=%v", key, value))
+			}
+		}
+	}
+	cmd.Env = env
+
+	// Execute command
+	output, err := cmd.CombinedOutput()
+	if err != nil {
+		// Check if it was cancelled
+		if ctx.Err() != nil {
+			work.Execution.Warn("Command cancelled: %s", ctx.Err().Error())
+			work.Execution.Status = "cancelled"
+		} else {
+			work.Execution.Error("Command failed: %s, output: %s", err.Error(), string(output))
+			work.Execution.Status = "failed"
+
+			// Store error output
+			if len(output) > 0 {
+				errorInfo := map[string]interface{}{
+					"error":  err.Error(),
+					"output": string(output),
+				}
+				if errorBytes, jsonErr := jsoniter.Marshal(errorInfo); jsonErr == nil {
+					work.Execution.ErrorInfo = (*json.RawMessage)(&errorBytes)
+				}
+			}
+		}
+
+		// Save execution status
+		if saveErr := SaveExecution(work.Execution); saveErr != nil {
+			work.Execution.Error("Failed to save execution error: %s", saveErr.Error())
+		}
+
+		if ctx.Err() != nil {
+			return ctx.Err()
+		}
+		return fmt.Errorf("command execution failed: %v", err)
+	}
+
+	work.Execution.Info("Command completed successfully, output: %s", string(output))
+
+	// Update execution with success result
+	work.Execution.Status = "completed"
+	work.Execution.Progress = 100
+	if len(output) > 0 {
+		result := map[string]interface{}{
+			"output": string(output),
+		}
+		if resultBytes, err := jsoniter.Marshal(result); err == nil {
+			work.Execution.Result = (*json.RawMessage)(&resultBytes)
+		}
+	}
+
+	if saveErr := SaveExecution(work.Execution); saveErr != nil {
+		work.Execution.Error("Failed to save execution result: %s", saveErr.Error())
+		return fmt.Errorf("failed to save execution result: %w", saveErr)
+	}
+
+	return nil
+}
+
+// convertArgsForYaoRun converts arguments to proper format for yao run command
+func convertArgsForYaoRun(args []interface{}) []string {
+	result := make([]string, 0, len(args))
+
+	for _, arg := range args {
+		if arg == nil {
+			result = append(result, "")
+			continue
+		}
+
+		// Use type assertion for basic types - direct conversion
+		switch v := arg.(type) {
+		case string:
+			result = append(result, v)
+		case bool:
+			result = append(result, fmt.Sprintf("%t", v))
+		case int, int8, int16, int32, int64, uint, uint8, uint16, uint32, uint64:
+			result = append(result, fmt.Sprintf("%d", v))
+		case float32, float64:
+			result = append(result, fmt.Sprintf("%g", v))
+		default:
+			// Complex types (slice, map, struct, etc.) need JSON serialization with :: prefix
+			if argBytes, err := jsoniter.Marshal(arg); err == nil {
+				result = append(result, "::"+string(argBytes))
+			} else {
+				// Fallback to string representation if JSON marshaling fails
+				result = append(result, fmt.Sprintf("%v", arg))
+			}
+		}
+	}
+
+	return result
+}

job/progress_test.go

@@ -1,261 +0,0 @@
-package job_test
-
-import (
-	"context"
-	"fmt"
-	"testing"
-	"time"
-
-	"github.com/yaoapp/gou/model"
-	"github.com/yaoapp/yao/config"
-	"github.com/yaoapp/yao/job"
-	"github.com/yaoapp/yao/test"
-)
-
-// TestProgressManager tests progress management functionality
-func TestProgressManager(t *testing.T) {
-	test.Prepare(t, config.Conf)
-	defer test.Clean()
-
-	// Create test job
-	testJob, err := job.Once(job.GOROUTINE, map[string]interface{}{
-		"name": "Test Progress Job",
-	})
-	if err != nil {
-		t.Fatalf("Failed to create test job: %v", err)
-	}
-
-	// Get progress manager
-	progressManager := testJob.Progress()
-	if progressManager == nil {
-		t.Fatal("Failed to get progress manager")
-	}
-
-	// Test initial progress
-	err = progressManager.Set(0, "Starting")
-	if err != nil {
-		t.Fatalf("Failed to set initial progress: %v", err)
-	}
-
-	// Test progress updates
-	err = progressManager.Set(25, "25% complete")
-	if err != nil {
-		t.Fatalf("Failed to set progress to 25: %v", err)
-	}
-
-	err = progressManager.Set(50, "50% complete")
-	if err != nil {
-		t.Fatalf("Failed to set progress to 50: %v", err)
-	}
-
-	err = progressManager.Set(75, "75% complete")
-	if err != nil {
-		t.Fatalf("Failed to set progress to 75: %v", err)
-	}
-
-	err = progressManager.Set(100, "Complete")
-	if err != nil {
-		t.Fatalf("Failed to set progress to 100: %v", err)
-	}
-}
-
-// TestProgressWithExecution tests progress updates during job execution
-func TestProgressWithExecution(t *testing.T) {
-	test.Prepare(t, config.Conf)
-	defer test.Clean()
-
-	// Create test job
-	testJob, err := job.Once(job.GOROUTINE, map[string]interface{}{
-		"name": "Test Progress Execution Job",
-	})
-	if err != nil {
-		t.Fatalf("Failed to create test job: %v", err)
-	}
-
-	// Channel to signal completion
-	done := make(chan int, 1)
-
-	// Progress tracking handler
-	progressHandler := func(ctx context.Context, execution *job.Execution) error {
-		// Test SetProgress method on execution
-		for i := 0; i <= 100; i += 20 {
-			err := execution.SetProgress(i, fmt.Sprintf("Progress: %d%%", i))
-			if err != nil {
-				return err
-			}
-			time.Sleep(50 * time.Millisecond)
-		}
-		done <- 100
-		return nil
-	}
-
-	err = testJob.Add(1, progressHandler)
-	if err != nil {
-		t.Fatalf("Failed to add progress handler: %v", err)
-	}
-
-	// Start worker manager
-	wm := job.NewWorkerManagerForTest(2)
-	wm.Start()
-	defer wm.Stop()
-	testJob.SetWorkerManager(wm)
-
-	// Start job
-	err = testJob.Start()
-	if err != nil {
-		t.Fatalf("Failed to start job: %v", err)
-	}
-
-	// Wait for job completion or timeout
-	var finalProgress int
-	select {
-	case finalProgress = <-done:
-		t.Logf("Job completed with progress: %d", finalProgress)
-	case <-time.After(10 * time.Second):
-		t.Error("Job execution timeout")
-		return
-	}
-
-	// Give some extra time for database operations to complete
-	time.Sleep(200 * time.Millisecond)
-
-	// Check final execution state
-	executions, err := testJob.GetExecutions()
-	if err != nil {
-		t.Fatalf("Failed to get executions: %v", err)
-	}
-	if len(executions) == 0 {
-		t.Error("Expected at least one execution")
-		return
-	}
-
-	execution := executions[0]
-
-	// Check final progress - we know it completed with 100 from the handler
-	if finalProgress != 100 {
-		t.Errorf("Expected handler to complete with 100, got %d", finalProgress)
-	}
-
-	// The database might not have the latest progress due to async operations
-	t.Logf("Final execution progress in database: %d", execution.Progress)
-	t.Logf("Final progress from handler: %d", finalProgress)
-}
-
-// TestProgressWithDatabase tests progress persistence in database
-func TestProgressWithDatabase(t *testing.T) {
-	test.Prepare(t, config.Conf)
-	defer test.Clean()
-
-	// Create test job
-	testJob, err := job.Once(job.GOROUTINE, map[string]interface{}{
-		"name": "Test Progress Database Job",
-	})
-	if err != nil {
-		t.Fatalf("Failed to create test job: %v", err)
-	}
-
-	// Add handler to save job first
-	err = testJob.Add(1, func(ctx context.Context, execution *job.Execution) error {
-		return nil
-	})
-	if err != nil {
-		t.Fatalf("Failed to add handler: %v", err)
-	}
-
-	// Create execution manually to test progress persistence
-	testExecution := &job.Execution{
-		ExecutionID:     "test-progress-exec-001",
-		JobID:           testJob.JobID,
-		Status:          "running",
-		TriggerCategory: "manual",
-		Progress:        0,
-		CreatedAt:       time.Now(),
-		UpdatedAt:       time.Now(),
-	}
-
-	err = job.SaveExecution(testExecution)
-	if err != nil {
-		t.Fatalf("Failed to save test execution: %v", err)
-	}
-
-	// Test progress updates through SetProgress
-	progressValues := []int{10, 25, 50, 75, 90, 100}
-	for _, progress := range progressValues {
-		testExecution.Progress = progress
-		err = testExecution.SetProgress(progress, fmt.Sprintf("Progress: %d%%", progress))
-		if err != nil {
-			t.Fatalf("Failed to set progress to %d: %v", progress, err)
-		}
-
-		// Verify progress was saved to database
-		savedExecution, err := job.GetExecution(testExecution.ExecutionID, model.QueryParam{})
-		if err != nil {
-			t.Fatalf("Failed to get saved execution: %v", err)
-		}
-
-		if savedExecution.Progress != progress {
-			t.Errorf("Expected saved progress %d, got %d", progress, savedExecution.Progress)
-		}
-	}
-
-	// Clean up
-	job.RemoveExecutions([]string{testExecution.ExecutionID})
-	job.RemoveJobs([]string{testJob.JobID})
-}
-
-// TestGetProgress tests live progress retrieval
-func TestGetProgress(t *testing.T) {
-	test.Prepare(t, config.Conf)
-	defer test.Clean()
-
-	// Create test execution
-	testExecution := &job.Execution{
-		ExecutionID:     "test-get-progress-001",
-		JobID:           "test-job-001",
-		Status:          "running",
-		TriggerCategory: "manual",
-		Progress:        75,
-		CreatedAt:       time.Now(),
-		UpdatedAt:       time.Now(),
-	}
-
-	err := job.SaveExecution(testExecution)
-	if err != nil {
-		t.Fatalf("Failed to save test execution: %v", err)
-	}
-
-	// Test GetProgress function
-	callbackCalled := false
-	progress, err := job.GetProgress(testExecution.ExecutionID, func(p *job.Progress) {
-		callbackCalled = true
-		if p.ExecutionID != testExecution.ExecutionID {
-			t.Errorf("Expected execution ID %s, got %s", testExecution.ExecutionID, p.ExecutionID)
-		}
-		if p.Progress != 75 {
-			t.Errorf("Expected progress 75, got %d", p.Progress)
-		}
-	})
-
-	if err != nil {
-		t.Fatalf("Failed to get progress: %v", err)
-	}
-
-	if progress == nil {
-		t.Fatal("Expected progress object, got nil")
-	}
-
-	if progress.ExecutionID != testExecution.ExecutionID {
-		t.Errorf("Expected execution ID %s, got %s", testExecution.ExecutionID, progress.ExecutionID)
-	}
-
-	if progress.Progress != 75 {
-		t.Errorf("Expected progress 75, got %d", progress.Progress)
-	}
-
-	if !callbackCalled {
-		t.Error("Expected callback to be called")
-	}
-
-	// Clean up
-	job.RemoveExecutions([]string{testExecution.ExecutionID})
-}

job/types.go

@@ -3,6 +3,7 @@ package job
 import (
 	"context"
 	"encoding/json"
+	"sync"
 	"time"
 )
 
@@ -51,8 +52,59 @@ const (
 	Trace
 )
 
-// HandlerFunc the job handler function
+// ExecutionType represents different execution methods
-type HandlerFunc func(ctx context.Context, execution *Execution) error
+type ExecutionType string
+
+// Execution type constants
+const (
+	ExecutionTypeProcess ExecutionType = "process" // Yao process (default)
+	ExecutionTypeCommand ExecutionType = "command" // System command
+)
+
+// ExecutionOptions holds common execution options
+type ExecutionOptions struct {
+	Priority   int                    `json:"priority"`    // Execution priority (higher = more important)
+	SharedData map[string]interface{} `json:"shared_data"` // Shared data (session, context, etc.)
+}
+
+// NewExecutionOptions creates a new ExecutionOptions with default values
+func NewExecutionOptions() *ExecutionOptions {
+	return &ExecutionOptions{
+		Priority:   0,
+		SharedData: make(map[string]interface{}),
+	}
+}
+
+// WithPriority sets the priority and returns the options for chaining
+func (o *ExecutionOptions) WithPriority(priority int) *ExecutionOptions {
+	o.Priority = priority
+	return o
+}
+
+// WithSharedData sets shared data and returns the options for chaining
+func (o *ExecutionOptions) WithSharedData(data map[string]interface{}) *ExecutionOptions {
+	o.SharedData = data
+	return o
+}
+
+// AddSharedData adds a key-value pair to shared data and returns the options for chaining
+func (o *ExecutionOptions) AddSharedData(key string, value interface{}) *ExecutionOptions {
+	if o.SharedData == nil {
+		o.SharedData = make(map[string]interface{})
+	}
+	o.SharedData[key] = value
+	return o
+}
+
+// ExecutionConfig holds execution configuration based on type
+type ExecutionConfig struct {
+	Type        ExecutionType     `json:"type"`
+	ProcessName string            `json:"process_name,omitempty"` // Yao process name
+	ProcessArgs []interface{}     `json:"process_args,omitempty"` // Yao process arguments
+	Command     string            `json:"command,omitempty"`      // System command
+	CommandArgs []string          `json:"command_args,omitempty"` // Command arguments
+	Environment map[string]string `json:"environment,omitempty"`  // Environment variables
+}
 
 // Job represents the main job entity
 type Job struct {
@@ -87,9 +139,12 @@ type Job struct {
 	Executions []Execution `json:"executions,omitempty"`
 	Logs       []Log       `json:"logs,omitempty"`
 
-	ctx           context.Context
+	ctx    context.Context
-	cancel        context.CancelFunc
+	cancel context.CancelFunc
-	workerManager *WorkerManager // For testing: allows using custom worker manager
+
+	// Job-level cancellation for running executions
+	executionContexts map[string]context.CancelFunc // executionID -> cancel function
+	executionMutex    sync.RWMutex
 }
 
 // Category represents job categories for organization
@@ -112,31 +167,33 @@ type Category struct {
 
 // Execution represents individual job execution instances
 type Execution struct {
-	ID                uint             `json:"id"`
+	ID                uint              `json:"id"`
-	ExecutionID       string           `json:"execution_id"`
+	ExecutionID       string            `json:"execution_id"`
-	JobID             string           `json:"job_id"`
+	JobID             string            `json:"job_id"`
-	Status            string           `json:"status"` // default: "queued"
+	Status            string            `json:"status"` // default: "queued"
-	TriggerCategory   string           `json:"trigger_category"`
+	TriggerCategory   string            `json:"trigger_category"`
-	TriggerSource     *string          `json:"trigger_source,omitempty"`      // nullable: true
+	TriggerSource     *string           `json:"trigger_source,omitempty"`      // nullable: true
-	TriggerContext    *json.RawMessage `json:"trigger_context,omitempty"`     // nullable: true
+	TriggerContext    *json.RawMessage  `json:"trigger_context,omitempty"`     // nullable: true
-	ScheduledAt       *time.Time       `json:"scheduled_at,omitempty"`        // nullable: true
+	ScheduledAt       *time.Time        `json:"scheduled_at,omitempty"`        // nullable: true
-	WorkerID          *string          `json:"worker_id,omitempty"`           // nullable: true
+	WorkerID          *string           `json:"worker_id,omitempty"`           // nullable: true
-	ProcessID         *string          `json:"process_id,omitempty"`          // nullable: true
+	ProcessID         *string           `json:"process_id,omitempty"`          // nullable: true
-	RetryAttempt      int              `json:"retry_attempt"`                 // default: 0
+	RetryAttempt      int               `json:"retry_attempt"`                 // default: 0
-	ParentExecutionID *string          `json:"parent_execution_id,omitempty"` // nullable: true
+	ParentExecutionID *string           `json:"parent_execution_id,omitempty"` // nullable: true
-	StartedAt         *time.Time       `json:"started_at,omitempty"`          // nullable: true
+	StartedAt         *time.Time        `json:"started_at,omitempty"`          // nullable: true
-	EndedAt           *time.Time       `json:"ended_at,omitempty"`            // nullable: true
+	EndedAt           *time.Time        `json:"ended_at,omitempty"`            // nullable: true
-	TimeoutSeconds    *int             `json:"timeout_seconds,omitempty"`     // nullable: true
+	TimeoutSeconds    *int              `json:"timeout_seconds,omitempty"`     // nullable: true
-	Duration          *int             `json:"duration,omitempty"`            // nullable: true
+	Duration          *int              `json:"duration,omitempty"`            // nullable: true
-	Progress          int              `json:"progress"`                      // default: 0
+	Progress          int               `json:"progress"`                      // default: 0
-	ConfigSnapshot    *json.RawMessage `json:"config_snapshot,omitempty"`     // nullable: true
+	ExecutionConfig   *ExecutionConfig  `json:"execution_config,omitempty"`    // Execution configuration
-	Result            *json.RawMessage `json:"result,omitempty"`              // nullable: true
+	ExecutionOptions  *ExecutionOptions `json:"execution_options,omitempty"`   // Execution options (priority, shared data)
-	ErrorInfo         *json.RawMessage `json:"error_info,omitempty"`          // nullable: true
+	ConfigSnapshot    *json.RawMessage  `json:"config_snapshot,omitempty"`     // nullable: true
-	StackTrace        *string          `json:"stack_trace,omitempty"`         // nullable: true
+	Result            *json.RawMessage  `json:"result,omitempty"`              // nullable: true
-	Metrics           *json.RawMessage `json:"metrics,omitempty"`             // nullable: true
+	ErrorInfo         *json.RawMessage  `json:"error_info,omitempty"`          // nullable: true
-	Context           *json.RawMessage `json:"context,omitempty"`             // nullable: true
+	StackTrace        *string           `json:"stack_trace,omitempty"`         // nullable: true
-	CreatedAt         time.Time        `json:"created_at"`
+	Metrics           *json.RawMessage  `json:"metrics,omitempty"`             // nullable: true
-	UpdatedAt         time.Time        `json:"updated_at"`
+	Context           *json.RawMessage  `json:"context,omitempty"`             // nullable: true
+	CreatedAt         time.Time         `json:"created_at"`
+	UpdatedAt         time.Time         `json:"updated_at"`
 
 	// Relationships
 	Job             *Job        `json:"job,omitempty"`

job/types_test.go

@@ -1,334 +0,0 @@
-package job_test
-
-import (
-	"context"
-	"encoding/json"
-	"fmt"
-	"testing"
-	"time"
-
-	"github.com/yaoapp/yao/job"
-)
-
-// TestJobTypes tests job type definitions and constants
-func TestJobTypes(t *testing.T) {
-	// Test ModeType constants
-	if job.GOROUTINE != "GOROUTINE" {
-		t.Errorf("Expected GOROUTINE mode to be 'GOROUTINE', got '%s'", job.GOROUTINE)
-	}
-	if job.PROCESS != "PROCESS" {
-		t.Errorf("Expected PROCESS mode to be 'PROCESS', got '%s'", job.PROCESS)
-	}
-
-	// Test ScheduleType constants
-	if job.ScheduleTypeOnce != "once" {
-		t.Errorf("Expected ScheduleTypeOnce to be 'once', got '%s'", job.ScheduleTypeOnce)
-	}
-	if job.ScheduleTypeCron != "cron" {
-		t.Errorf("Expected ScheduleTypeCron to be 'cron', got '%s'", job.ScheduleTypeCron)
-	}
-	if job.ScheduleTypeDaemon != "daemon" {
-		t.Errorf("Expected ScheduleTypeDaemon to be 'daemon', got '%s'", job.ScheduleTypeDaemon)
-	}
-
-	// Test LogLevel constants
-	expectedLevels := map[job.LogLevel]string{
-		job.Debug: "debug",
-		job.Info:  "info",
-		job.Warn:  "warn",
-		job.Error: "error",
-		job.Fatal: "fatal",
-		job.Panic: "panic",
-		job.Trace: "trace",
-	}
-
-	for level, name := range expectedLevels {
-		if level > 6 {
-			t.Errorf("LogLevel %s has invalid value %d", name, level)
-		}
-	}
-}
-
-// TestJobStructure tests Job struct serialization and deserialization
-func TestJobStructure(t *testing.T) {
-	// Create a test job
-	now := time.Now()
-	description := "Test job description"
-	timeout := 300
-	nextRun := now.Add(time.Hour)
-	lastRun := now.Add(-time.Hour)
-	currentExecID := "exec-123"
-
-	testJob := &job.Job{
-		ID:                 1,
-		JobID:              "test-job-001",
-		Name:               "Test Job",
-		Icon:               &description,
-		Description:        &description,
-		CategoryID:         "test-category",
-		MaxWorkerNums:      2,
-		Status:             "ready",
-		Mode:               job.GOROUTINE,
-		ScheduleType:       string(job.ScheduleTypeOnce),
-		ScheduleExpression: nil,
-		MaxRetryCount:      3,
-		DefaultTimeout:     &timeout,
-		Priority:           5,
-		CreatedBy:          "test-user",
-		NextRunAt:          &nextRun,
-		LastRunAt:          &lastRun,
-		CurrentExecutionID: &currentExecID,
-		Config:             map[string]interface{}{"key": "value"},
-		Sort:               1,
-		Enabled:            true,
-		System:             false,
-		Readonly:           false,
-		CreatedAt:          now,
-		UpdatedAt:          now,
-	}
-
-	// Test JSON serialization
-	jsonData, err := json.Marshal(testJob)
-	if err != nil {
-		t.Fatalf("Failed to marshal job to JSON: %v", err)
-	}
-
-	// Test JSON deserialization
-	var deserializedJob job.Job
-	err = json.Unmarshal(jsonData, &deserializedJob)
-	if err != nil {
-		t.Fatalf("Failed to unmarshal job from JSON: %v", err)
-	}
-
-	// Verify key fields
-	if deserializedJob.JobID != testJob.JobID {
-		t.Errorf("Expected JobID '%s', got '%s'", testJob.JobID, deserializedJob.JobID)
-	}
-	if deserializedJob.Name != testJob.Name {
-		t.Errorf("Expected Name '%s', got '%s'", testJob.Name, deserializedJob.Name)
-	}
-	if deserializedJob.Mode != testJob.Mode {
-		t.Errorf("Expected Mode '%s', got '%s'", testJob.Mode, deserializedJob.Mode)
-	}
-	if deserializedJob.Priority != testJob.Priority {
-		t.Errorf("Expected Priority %d, got %d", testJob.Priority, deserializedJob.Priority)
-	}
-}
-
-// TestCategoryStructure tests Category struct
-func TestCategoryStructure(t *testing.T) {
-	now := time.Now()
-	description := "Test category description"
-
-	testCategory := &job.Category{
-		ID:          1,
-		CategoryID:  "test-category-001",
-		Name:        "Test Category",
-		Icon:        &description,
-		Description: &description,
-		Sort:        1,
-		System:      false,
-		Enabled:     true,
-		Readonly:    false,
-		CreatedAt:   now,
-		UpdatedAt:   now,
-	}
-
-	// Test JSON serialization
-	jsonData, err := json.Marshal(testCategory)
-	if err != nil {
-		t.Fatalf("Failed to marshal category to JSON: %v", err)
-	}
-
-	// Test JSON deserialization
-	var deserializedCategory job.Category
-	err = json.Unmarshal(jsonData, &deserializedCategory)
-	if err != nil {
-		t.Fatalf("Failed to unmarshal category from JSON: %v", err)
-	}
-
-	// Verify key fields
-	if deserializedCategory.CategoryID != testCategory.CategoryID {
-		t.Errorf("Expected CategoryID '%s', got '%s'", testCategory.CategoryID, deserializedCategory.CategoryID)
-	}
-	if deserializedCategory.Name != testCategory.Name {
-		t.Errorf("Expected Name '%s', got '%s'", testCategory.Name, deserializedCategory.Name)
-	}
-	if deserializedCategory.Enabled != testCategory.Enabled {
-		t.Errorf("Expected Enabled %v, got %v", testCategory.Enabled, deserializedCategory.Enabled)
-	}
-}
-
-// TestExecutionStructure tests Execution struct
-func TestExecutionStructure(t *testing.T) {
-	now := time.Now()
-	startedAt := now.Add(-time.Minute)
-	endedAt := now
-	timeout := 300
-	duration := 60000
-	parentExecID := "parent-exec-001"
-	workerID := "worker-001"
-	processID := "process-001"
-
-	testExecution := &job.Execution{
-		ID:                1,
-		ExecutionID:       "test-execution-001",
-		JobID:             "test-job-001",
-		Status:            "completed",
-		TriggerCategory:   "manual",
-		TriggerSource:     &workerID,
-		ScheduledAt:       &startedAt,
-		WorkerID:          &workerID,
-		ProcessID:         &processID,
-		RetryAttempt:      0,
-		ParentExecutionID: &parentExecID,
-		StartedAt:         &startedAt,
-		EndedAt:           &endedAt,
-		TimeoutSeconds:    &timeout,
-		Duration:          &duration,
-		Progress:          100,
-		CreatedAt:         now,
-		UpdatedAt:         now,
-	}
-
-	// Test JSON serialization
-	jsonData, err := json.Marshal(testExecution)
-	if err != nil {
-		t.Fatalf("Failed to marshal execution to JSON: %v", err)
-	}
-
-	// Test JSON deserialization
-	var deserializedExecution job.Execution
-	err = json.Unmarshal(jsonData, &deserializedExecution)
-	if err != nil {
-		t.Fatalf("Failed to unmarshal execution from JSON: %v", err)
-	}
-
-	// Verify key fields
-	if deserializedExecution.ExecutionID != testExecution.ExecutionID {
-		t.Errorf("Expected ExecutionID '%s', got '%s'", testExecution.ExecutionID, deserializedExecution.ExecutionID)
-	}
-	if deserializedExecution.JobID != testExecution.JobID {
-		t.Errorf("Expected JobID '%s', got '%s'", testExecution.JobID, deserializedExecution.JobID)
-	}
-	if deserializedExecution.Status != testExecution.Status {
-		t.Errorf("Expected Status '%s', got '%s'", testExecution.Status, deserializedExecution.Status)
-	}
-	if deserializedExecution.Progress != testExecution.Progress {
-		t.Errorf("Expected Progress %d, got %d", testExecution.Progress, deserializedExecution.Progress)
-	}
-}
-
-// TestLogStructure tests Log struct
-func TestLogStructure(t *testing.T) {
-	now := time.Now()
-	executionID := "test-execution-001"
-	source := "test-handler"
-	step := "initialization"
-	progress := 50
-	duration := 1000
-	errorCode := "ERR001"
-	stackTrace := "stack trace here"
-	workerID := "worker-001"
-	processID := "process-001"
-
-	testLog := &job.Log{
-		ID:          1,
-		JobID:       "test-job-001",
-		Level:       "info",
-		Message:     "Test log message",
-		Source:      &source,
-		ExecutionID: &executionID,
-		Step:        &step,
-		Progress:    &progress,
-		Duration:    &duration,
-		ErrorCode:   &errorCode,
-		StackTrace:  &stackTrace,
-		WorkerID:    &workerID,
-		ProcessID:   &processID,
-		Timestamp:   now,
-		Sequence:    1,
-		CreatedAt:   now,
-		UpdatedAt:   now,
-	}
-
-	// Test JSON serialization
-	jsonData, err := json.Marshal(testLog)
-	if err != nil {
-		t.Fatalf("Failed to marshal log to JSON: %v", err)
-	}
-
-	// Test JSON deserialization
-	var deserializedLog job.Log
-	err = json.Unmarshal(jsonData, &deserializedLog)
-	if err != nil {
-		t.Fatalf("Failed to unmarshal log from JSON: %v", err)
-	}
-
-	// Verify key fields
-	if deserializedLog.JobID != testLog.JobID {
-		t.Errorf("Expected JobID '%s', got '%s'", testLog.JobID, deserializedLog.JobID)
-	}
-	if deserializedLog.Level != testLog.Level {
-		t.Errorf("Expected Level '%s', got '%s'", testLog.Level, deserializedLog.Level)
-	}
-	if deserializedLog.Message != testLog.Message {
-		t.Errorf("Expected Message '%s', got '%s'", testLog.Message, deserializedLog.Message)
-	}
-	if deserializedLog.Sequence != testLog.Sequence {
-		t.Errorf("Expected Sequence %d, got %d", testLog.Sequence, deserializedLog.Sequence)
-	}
-}
-
-// TestProgressStructure tests Progress struct
-func TestProgressStructure(t *testing.T) {
-	testProgress := &job.Progress{
-		ExecutionID: "test-execution-001",
-		Progress:    75,
-		Message:     "75% complete",
-	}
-
-	// Test JSON serialization
-	jsonData, err := json.Marshal(testProgress)
-	if err != nil {
-		t.Fatalf("Failed to marshal progress to JSON: %v", err)
-	}
-
-	// Test JSON deserialization
-	var deserializedProgress job.Progress
-	err = json.Unmarshal(jsonData, &deserializedProgress)
-	if err != nil {
-		t.Fatalf("Failed to unmarshal progress from JSON: %v", err)
-	}
-
-	// Verify fields
-	if deserializedProgress.ExecutionID != testProgress.ExecutionID {
-		t.Errorf("Expected ExecutionID '%s', got '%s'", testProgress.ExecutionID, deserializedProgress.ExecutionID)
-	}
-	if deserializedProgress.Progress != testProgress.Progress {
-		t.Errorf("Expected Progress %d, got %d", testProgress.Progress, deserializedProgress.Progress)
-	}
-	if deserializedProgress.Message != testProgress.Message {
-		t.Errorf("Expected Message '%s', got '%s'", testProgress.Message, deserializedProgress.Message)
-	}
-}
-
-// TestHandlerFunc tests HandlerFunc type
-func TestHandlerFunc(t *testing.T) {
-	// Test handler function signature
-	var handler job.HandlerFunc = func(ctx context.Context, execution *job.Execution) error {
-		if ctx == nil {
-			return fmt.Errorf("context is nil")
-		}
-		if execution == nil {
-			return fmt.Errorf("execution is nil")
-		}
-		execution.Info("Handler executed successfully")
-		return nil
-	}
-
-	// Test handler function signature
-	if handler == nil {
-		t.Error("Handler function should not be nil")
-	}
-}

job/worker.go

@@ -4,8 +4,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"os"
-	"os/exec"
 	"runtime"
 	"sync"
 	"time"
@@ -40,7 +38,6 @@ type Worker struct {
 type WorkRequest struct {
 	Job       *Job
 	Execution *Execution
-	Handler   HandlerFunc
 	Context   context.Context
 }
 
@@ -48,14 +45,29 @@ type WorkRequest struct {
 var globalWorkerManager *WorkerManager
 var workerManagerOnce sync.Once
 
+// init initializes the global worker manager
+func init() {
+	// Start the global worker manager on package initialization
+	wm := GetWorkerManager()
+	wm.Start()
+}
+
 // GetWorkerManager returns the global worker manager instance
 func GetWorkerManager() *WorkerManager {
 	workerManagerOnce.Do(func() {
-		globalWorkerManager = NewWorkerManager(runtime.NumCPU() * 2) // Default to 2x CPU cores
+		globalWorkerManager = NewWorkerManager(getDefaultMaxWorkers()) // Use configurable default
 	})
 	return globalWorkerManager
 }
 
+// getDefaultMaxWorkers returns the default max workers count
+// This can be configured via environment variables or config files
+func getDefaultMaxWorkers() int {
+	// Use CPU count * 4 as default for optimal concurrency
+	// This provides good balance between resource utilization and system load
+	return runtime.NumCPU() * 4
+}
+
 // NewWorkerManagerForTest creates a new worker manager for testing (not singleton)
 func NewWorkerManagerForTest(maxWorkers int) *WorkerManager {
 	return NewWorkerManager(maxWorkers)
@@ -66,7 +78,7 @@ func NewWorkerManager(maxWorkers int) *WorkerManager {
 	return &WorkerManager{
 		maxWorkers:    maxWorkers,
 		activeWorkers: make(map[string]*Worker),
-		workQueue:     make(chan *WorkRequest, maxWorkers*2), // Buffer for queue
+		workQueue:     make(chan *WorkRequest, maxWorkers*4), // Allow 200% overload (4x buffer)
 		workerPool:    make(chan chan *WorkRequest, maxWorkers),
 		quit:          make(chan bool),
 	}
@@ -122,41 +134,36 @@ func (wm *WorkerManager) Stop() {
 	log.Info("Worker manager stopped")
 }
 
-// SubmitJob submits a job for execution
+// SubmitJob submits a job execution for processing with context (non-blocking)
-func (wm *WorkerManager) SubmitJob(job *Job, handler HandlerFunc) error {
+func (wm *WorkerManager) SubmitJob(ctx context.Context, job *Job, execution *Execution) error {
-	// Create execution record
+	// Check queue capacity before submitting
-	execution := &Execution{
+	queueLen := len(wm.workQueue)
-		ExecutionID:     uuid.New().String(),
+	queueCap := cap(wm.workQueue)
-		JobID:           job.JobID,
-		Status:          "queued",
-		TriggerCategory: "manual", // Default trigger
-		RetryAttempt:    0,
-		Progress:        0,
-		CreatedAt:       time.Now(),
-		UpdatedAt:       time.Now(),
-	}
 
-	// Save execution to database
+	// Allow reasonable backlog but prevent unlimited accumulation
-	if err := SaveExecution(execution); err != nil {
+	// Reject only when queue is completely full to maximize throughput
-		return fmt.Errorf("failed to save execution: %w", err)
+	if queueLen >= queueCap {
+		return fmt.Errorf("work queue is full (%d/%d), please retry later", queueLen, queueCap)
 	}
 
 	// Create work request
 	workRequest := &WorkRequest{
 		Job:       job,
 		Execution: execution,
-		Handler:   handler,
+		Context:   ctx,
-		Context:   context.Background(),
 	}
 
-	// Submit to work queue
+	// Submit asynchronously to avoid blocking
-	select {
+	go func() {
-	case wm.workQueue <- workRequest:
+		select {
-		log.Debug("Job %s submitted to work queue", job.JobID)
+		case wm.workQueue <- workRequest:
-		return nil
+			log.Debug("Job %s execution %s submitted to work queue", job.JobID, execution.ExecutionID)
-	default:
+		case <-ctx.Done():
-		return fmt.Errorf("work queue is full")
+			log.Warn("Job %s execution %s submission cancelled", job.JobID, execution.ExecutionID)
-	}
+		}
+	}()
+
+	return nil
 }
 
 // dispatch dispatches work requests to available workers
@@ -185,6 +192,11 @@ func (wm *WorkerManager) GetActiveWorkers() int {
 	return len(wm.activeWorkers)
 }
 
+// GetQueueStatus returns queue length and capacity for monitoring
+func (wm *WorkerManager) GetQueueStatus() (length int, capacity int) {
+	return len(wm.workQueue), cap(wm.workQueue)
+}
+
 // NewWorker creates a new worker
 func NewWorker(workerPool chan chan *WorkRequest, mode ModeType) *Worker {
 	ctx, cancel := context.WithCancel(context.Background())
@@ -358,67 +370,60 @@ func (w *Worker) processWork(work *WorkRequest) {
 		log.Warn("Failed to save final job status (database may be closed): %v", err)
 	}
 
+	// Clean up execution context from job
+	if work.Job.executionContexts != nil {
+		work.Job.executionMutex.Lock()
+		delete(work.Job.executionContexts, work.Execution.ExecutionID)
+		work.Job.executionMutex.Unlock()
+	}
+
 	log.Debug("Worker %s finished processing job %s", w.ID, work.Job.JobID)
 }
 
 // executeInGoroutine executes job in goroutine mode
 func (w *Worker) executeInGoroutine(ctx context.Context, work *WorkRequest, progress *Progress) error {
-	// Execute handler directly in current goroutine
+	// Execute based on execution config type
-	return work.Handler(ctx, work.Execution)
+	if work.Execution.ExecutionConfig == nil {
+		return fmt.Errorf("execution config is nil")
+	}
+
+	// Create goroutine executor
+	goroutineExecutor := &Goroutine{}
+
+	switch work.Execution.ExecutionConfig.Type {
+	case ExecutionTypeProcess:
+		return goroutineExecutor.ExecuteYaoProcess(ctx, work, progress)
+
+	case ExecutionTypeCommand:
+		return goroutineExecutor.ExecuteSystemCommand(ctx, work, progress)
+
+	default:
+		return fmt.Errorf("unsupported execution type: %s", work.Execution.ExecutionConfig.Type)
+	}
 }
 
 // executeInProcess executes job in process mode
 func (w *Worker) executeInProcess(ctx context.Context, work *WorkRequest, progress *Progress) error {
-	// For process mode, we would typically spawn a separate process
+	// Execute based on execution config type using independent process
-	// For now, we'll simulate this with a goroutine but with process isolation concepts
+	if work.Execution.ExecutionConfig == nil {
+		return fmt.Errorf("execution config is nil")
+	}
 
-	// Set process ID (simulated)
+	// Set process ID (will be actual process ID)
 	processID := fmt.Sprintf("proc_%s", uuid.New().String()[:8])
 	work.Execution.ProcessID = &processID
 
-	// Create a separate goroutine to simulate process isolation
+	// Create process executor
-	done := make(chan error, 1)
+	processExecutor := &Process{}
 
-	go func() {
+	switch work.Execution.ExecutionConfig.Type {
-		defer func() {
+	case ExecutionTypeProcess:
-			if r := recover(); r != nil {
+		return processExecutor.ExecuteYaoProcess(ctx, work, progress)
-				done <- fmt.Errorf("process panic: %v", r)
-			}
-		}()
 
-		// Execute handler
+	case ExecutionTypeCommand:
-		err := work.Handler(ctx, work.Execution)
+		return processExecutor.ExecuteSystemCommand(ctx, work, progress)
-		done <- err
-	}()
 
-	// Wait for completion or timeout
+	default:
-	select {
+		return fmt.Errorf("unsupported execution type: %s", work.Execution.ExecutionConfig.Type)
-	case err := <-done:
-		return err
-	case <-ctx.Done():
-		return ctx.Err()
 	}
 }
-
-// executeInRealProcess executes job in a real separate process (future implementation)
-func (w *Worker) executeInRealProcess(ctx context.Context, work *WorkRequest, progress *Progress) error {
-	// This would be used for true process isolation
-	// For now, it's a placeholder for future implementation
-
-	// Create command to execute job in separate process
-	cmd := exec.CommandContext(ctx, os.Args[0], "job-execute", work.Execution.ExecutionID)
-
-	// Set environment variables
-	cmd.Env = append(os.Environ(),
-		fmt.Sprintf("JOB_ID=%s", work.Job.JobID),
-		fmt.Sprintf("EXECUTION_ID=%s", work.Execution.ExecutionID),
-	)
-
-	// Execute command
-	output, err := cmd.CombinedOutput()
-	if err != nil {
-		return fmt.Errorf("process execution failed: %v, output: %s", err, string(output))
-	}
-
-	return nil
-}

job/worker_test.go

@@ -1,320 +0,0 @@
-package job_test
-
-import (
-	"context"
-	"fmt"
-	"testing"
-	"time"
-
-	"github.com/yaoapp/yao/config"
-	"github.com/yaoapp/yao/job"
-	"github.com/yaoapp/yao/test"
-)
-
-// TestWorkerManagerLifecycle tests worker manager lifecycle
-func TestWorkerManagerLifecycle(t *testing.T) {
-	test.Prepare(t, config.Conf)
-	defer test.Clean()
-
-	// Create a new worker manager for testing (not singleton)
-	wm := job.NewWorkerManagerForTest(2)
-	if wm == nil {
-		t.Fatal("Failed to create worker manager")
-	}
-
-	// Test initial state
-	if wm.GetActiveWorkers() != 0 {
-		t.Error("Expected 0 active workers initially")
-	}
-
-	// Test start
-	wm.Start()
-	if wm.GetActiveWorkers() == 0 {
-		t.Error("Expected active workers after start")
-	}
-
-	// Test stop
-	wm.Stop()
-	// Note: Workers might still be active briefly after stop due to cleanup time
-	time.Sleep(100 * time.Millisecond)
-
-	// Test restart
-	wm.Start()
-	if wm.GetActiveWorkers() == 0 {
-		t.Error("Expected active workers after restart")
-	}
-
-	// Final cleanup
-	wm.Stop()
-}
-
-// TestWorkerJobSubmission tests job submission to workers
-func TestWorkerJobSubmission(t *testing.T) {
-	test.Prepare(t, config.Conf)
-	defer test.Clean()
-
-	// Create test job
-	testJob, err := job.Once(job.GOROUTINE, map[string]interface{}{
-		"name": "Test Worker Submission Job",
-	})
-	if err != nil {
-		t.Fatalf("Failed to create test job: %v", err)
-	}
-
-	// Test handler
-	executed := make(chan bool, 1)
-	testHandler := func(ctx context.Context, execution *job.Execution) error {
-		execution.Info("Test handler executed")
-		executed <- true
-		return nil
-	}
-
-	// Add handler and save job
-	err = testJob.Add(1, testHandler)
-	if err != nil {
-		t.Fatalf("Failed to add handler: %v", err)
-	}
-
-	// Create worker manager for testing
-	wm := job.NewWorkerManagerForTest(2)
-	wm.Start()
-	defer wm.Stop()
-
-	// Submit job
-	err = wm.SubmitJob(testJob, testHandler)
-	if err != nil {
-		t.Fatalf("Failed to submit job: %v", err)
-	}
-
-	// Wait for execution
-	select {
-	case <-executed:
-		t.Log("Job executed successfully")
-	case <-time.After(5 * time.Second):
-		t.Error("Job execution timeout")
-	}
-
-	// Check executions
-	executions, err := testJob.GetExecutions()
-	if err != nil {
-		t.Fatalf("Failed to get executions: %v", err)
-	}
-	if len(executions) == 0 {
-		t.Error("Expected at least one execution")
-	}
-}
-
-// TestWorkerModes tests different worker execution modes
-func TestWorkerModes(t *testing.T) {
-	test.Prepare(t, config.Conf)
-	defer test.Clean()
-
-	wm := job.NewWorkerManagerForTest(4)
-	wm.Start()
-	defer wm.Stop()
-
-	// Test GOROUTINE mode
-	goroutineJob, err := job.Once(job.GOROUTINE, map[string]interface{}{
-		"name": "Test Goroutine Mode Job",
-	})
-	if err != nil {
-		t.Fatalf("Failed to create goroutine job: %v", err)
-	}
-
-	goroutineExecuted := make(chan bool, 1)
-	goroutineHandler := func(ctx context.Context, execution *job.Execution) error {
-		if execution.Job != nil && execution.Job.Mode != job.GOROUTINE {
-			t.Errorf("Expected GOROUTINE mode, got %v", execution.Job.Mode)
-		}
-		goroutineExecuted <- true
-		return nil
-	}
-
-	err = goroutineJob.Add(1, goroutineHandler)
-	if err != nil {
-		t.Fatalf("Failed to add goroutine handler: %v", err)
-	}
-
-	goroutineJob.SetWorkerManager(wm)
-
-	err = goroutineJob.Start()
-	if err != nil {
-		t.Fatalf("Failed to start goroutine job: %v", err)
-	}
-
-	// Test PROCESS mode
-	processJob, err := job.Once(job.PROCESS, map[string]interface{}{
-		"name": "Test Process Mode Job",
-	})
-	if err != nil {
-		t.Fatalf("Failed to create process job: %v", err)
-	}
-
-	processExecuted := make(chan bool, 1)
-	processHandler := func(ctx context.Context, execution *job.Execution) error {
-		if execution.Job != nil && execution.Job.Mode != job.PROCESS {
-			t.Errorf("Expected PROCESS mode, got %v", execution.Job.Mode)
-		}
-		processExecuted <- true
-		return nil
-	}
-
-	err = processJob.Add(1, processHandler)
-	if err != nil {
-		t.Fatalf("Failed to add process handler: %v", err)
-	}
-
-	processJob.SetWorkerManager(wm)
-
-	err = processJob.Start()
-	if err != nil {
-		t.Fatalf("Failed to start process job: %v", err)
-	}
-
-	// Wait for both executions
-	timeout := time.After(10 * time.Second)
-	goroutineDone := false
-	processDone := false
-
-	for !goroutineDone || !processDone {
-		select {
-		case <-goroutineExecuted:
-			goroutineDone = true
-			t.Log("Goroutine job executed successfully")
-		case <-processExecuted:
-			processDone = true
-			t.Log("Process job executed successfully")
-		case <-timeout:
-			t.Error("Jobs execution timeout")
-			return
-		}
-	}
-
-	// Give extra time for database operations to complete
-	time.Sleep(500 * time.Millisecond)
-}
-
-// TestWorkerErrorHandling tests worker error handling
-func TestWorkerErrorHandling(t *testing.T) {
-	test.Prepare(t, config.Conf)
-	defer test.Clean()
-
-	// Create test job
-	errorJob, err := job.Once(job.GOROUTINE, map[string]interface{}{
-		"name": "Test Error Handling Job",
-	})
-	if err != nil {
-		t.Fatalf("Failed to create error job: %v", err)
-	}
-
-	// Error handler
-	errorHandler := func(ctx context.Context, execution *job.Execution) error {
-		execution.Error("Test error occurred")
-		return fmt.Errorf("intentional test error")
-	}
-
-	err = errorJob.Add(1, errorHandler)
-	if err != nil {
-		t.Fatalf("Failed to add error handler: %v", err)
-	}
-
-	wm := job.NewWorkerManagerForTest(4)
-	wm.Start()
-	defer wm.Stop()
-	errorJob.SetWorkerManager(wm)
-
-	err = errorJob.Start()
-	if err != nil {
-		t.Fatalf("Failed to start error job: %v", err)
-	}
-
-	// Wait for execution to complete
-	time.Sleep(2 * time.Second)
-
-	// Give extra time for database operations to complete
-	time.Sleep(500 * time.Millisecond)
-
-	// Check execution status
-	executions, err := errorJob.GetExecutions()
-	if err != nil {
-		t.Fatalf("Failed to get executions: %v", err)
-	}
-	if len(executions) == 0 {
-		t.Error("Expected at least one execution")
-		return
-	}
-
-	execution := executions[0]
-	if execution.Status != "failed" {
-		t.Errorf("Expected execution status 'failed', got '%s'", execution.Status)
-	}
-
-	if execution.ErrorInfo == nil {
-		t.Error("Expected error info to be set")
-	}
-}
-
-// TestWorkerConcurrency tests worker concurrency
-func TestWorkerConcurrency(t *testing.T) {
-	test.Prepare(t, config.Conf)
-	defer test.Clean()
-
-	wm := job.NewWorkerManagerForTest(4)
-	wm.Start()
-	defer wm.Stop()
-
-	const numJobs = 5
-	executed := make(chan bool, numJobs)
-
-	// Create and submit multiple jobs concurrently
-	for i := 0; i < numJobs; i++ {
-		testJob, err := job.Once(job.GOROUTINE, map[string]interface{}{
-			"name": fmt.Sprintf("Concurrent Test Job %d", i+1),
-		})
-		if err != nil {
-			t.Fatalf("Failed to create test job %d: %v", i+1, err)
-		}
-
-		jobID := i + 1
-		testHandler := func(ctx context.Context, execution *job.Execution) error {
-			execution.Info("Concurrent job %d executed", jobID)
-			time.Sleep(100 * time.Millisecond) // Simulate work
-			executed <- true
-			return nil
-		}
-
-		err = testJob.Add(1, testHandler)
-		if err != nil {
-			t.Fatalf("Failed to add handler for job %d: %v", i+1, err)
-		}
-
-		testJob.SetWorkerManager(wm)
-
-		err = testJob.Start()
-		if err != nil {
-			t.Fatalf("Failed to start job %d: %v", i+1, err)
-		}
-	}
-
-	// Wait for all jobs to complete
-	completed := 0
-	timeout := time.After(10 * time.Second)
-
-	for completed < numJobs {
-		select {
-		case <-executed:
-			completed++
-			t.Logf("Job %d completed", completed)
-		case <-timeout:
-			t.Errorf("Timeout: only %d out of %d jobs completed", completed, numJobs)
-			return
-		}
-	}
-
-	if completed != numJobs {
-		t.Errorf("Expected %d jobs to complete, got %d", numJobs, completed)
-	}
-
-	// Give extra time for database operations to complete
-	time.Sleep(500 * time.Millisecond)
-}

yao/models/job/execution.mod.yao

@@ -159,6 +159,15 @@
       "comment": "Execution duration in milliseconds",
       "nullable": true
     },
+    {
+      "name": "priority",
+      "type": "integer",
+      "label": "Priority",
+      "comment": "Execution priority (higher number = higher priority)",
+      "default": 0,
+      "nullable": false,
+      "index": true
+    },
     {
       "name": "progress",
       "type": "integer",
@@ -167,6 +176,13 @@
       "default": 0,
       "nullable": false
     },
+    {
+      "name": "execution_options",
+      "type": "json",
+      "label": "Execution Options",
+      "comment": "Execution options including priority and shared data",
+      "nullable": true
+    },
     {
       "name": "config_snapshot",
       "type": "json",