Enhance job management and category handling

- Introduced field lists for job, category, execution, and log queries to streamline data retrieval.
- Updated ListJobs, GetJob, GetCategories, and ListLogs functions to utilize new field lists for improved query efficiency.
- Enhanced job creation logic to set default values for enabled status, ensuring new jobs are active by default.
- Added category name resolution in job retrieval, enriching job data with associated category names.
- Improved error handling and logging throughout job and category management processes for better traceability.

job/data.go

@@ -12,6 +12,41 @@ import (
 	"github.com/yaoapp/xun/dbal"
 )
 
+// ========================
+// Field Lists for SELECT queries
+// ========================
+
+// JobFields defines the fields to select for job queries
+var JobFields = []interface{}{
+	"id", "job_id", "name", "icon", "description", "category_id",
+	"max_worker_nums", "status", "mode", "schedule_type", "schedule_expression",
+	"max_retry_count", "default_timeout", "priority", "created_by",
+	"next_run_at", "last_run_at", "current_execution_id", "config",
+	"sort", "enabled", "system", "readonly", "created_at", "updated_at",
+}
+
+// CategoryFields defines the fields to select for category queries
+var CategoryFields = []interface{}{
+	"id", "category_id", "name", "icon", "description",
+	"sort", "system", "enabled", "readonly", "created_at", "updated_at",
+}
+
+// ExecutionFields defines the fields to select for execution queries
+var ExecutionFields = []interface{}{
+	"id", "execution_id", "job_id", "status", "trigger_category", "trigger_source",
+	"trigger_context", "scheduled_at", "worker_id", "process_id", "retry_attempt",
+	"parent_execution_id", "started_at", "ended_at", "timeout_seconds", "duration",
+	"progress", "execution_config", "execution_options", "config_snapshot",
+	"result", "error_info", "stack_trace", "metrics", "context", "created_at", "updated_at",
+}
+
+// LogFields defines the fields to select for log queries
+var LogFields = []interface{}{
+	"id", "job_id", "level", "message", "context", "source", "execution_id",
+	"step", "progress", "duration", "error_code", "stack_trace",
+	"worker_id", "process_id", "timestamp", "sequence", "created_at", "updated_at",
+}
+
 // ========================
 // Jobs methods
 // ========================
@@ -22,7 +57,109 @@ func ListJobs(param model.QueryParam, page int, pagesize int) (maps.MapStrAny, e
 	if mod == nil {
 		return nil, fmt.Errorf("job model not found")
 	}
-	return mod.Paginate(param, page, pagesize)
+
+	// Set select fields if not already specified
+	if len(param.Select) == 0 {
+		param.Select = JobFields
+	}
+
+	// Debug logging
+	log.Debug("ListJobs called with param: %+v, page: %d, pagesize: %d", param, page, pagesize)
+
+	result, err := mod.Paginate(param, page, pagesize)
+	if err != nil {
+		log.Error("ListJobs query error: %v", err)
+		return nil, err
+	}
+
+	// Extract jobs data
+	log.Debug("ListJobs raw result: %+v", result)
+
+	jobsData, ok := result["data"].([]maps.MapStrAny)
+	if !ok {
+		log.Debug("Data type conversion failed, result[\"data\"] type: %T, value: %+v", result["data"], result["data"])
+		// Try alternative type conversion
+		if dataSlice, ok := result["data"].([]interface{}); ok {
+			jobsData = make([]maps.MapStrAny, len(dataSlice))
+			for i, item := range dataSlice {
+				if mapItem, ok := item.(maps.MapStrAny); ok {
+					jobsData[i] = mapItem
+				} else if mapItem, ok := item.(map[string]interface{}); ok {
+					jobsData[i] = maps.MapStrAny(mapItem)
+				} else {
+					log.Debug("Item %d type conversion failed: %T", i, item)
+					return result, nil
+				}
+			}
+		} else {
+			log.Debug("Alternative conversion also failed")
+			return result, nil
+		}
+	}
+
+	if len(jobsData) == 0 {
+		log.Debug("No jobs found in data")
+		return result, nil
+	}
+
+	// Collect unique category IDs
+	categoryIDs := make(map[string]bool)
+	for _, job := range jobsData {
+		if categoryID, exists := job["category_id"]; exists && categoryID != nil {
+			if categoryIDStr, ok := categoryID.(string); ok && categoryIDStr != "" {
+				categoryIDs[categoryIDStr] = true
+			}
+		}
+	}
+
+	// Query categories if we have category IDs
+	categoryMap := make(map[string]string)
+	if len(categoryIDs) > 0 {
+		categoryIDList := make([]string, 0, len(categoryIDs))
+		for categoryID := range categoryIDs {
+			categoryIDList = append(categoryIDList, categoryID)
+		}
+
+		categoryMod := model.Select("__yao.job.category")
+		if categoryMod != nil {
+			categoryParam := model.QueryParam{
+				Select: []interface{}{"category_id", "name"},
+				Wheres: []model.QueryWhere{
+					{Column: "category_id", OP: "in", Value: categoryIDList},
+				},
+			}
+
+			categories, err := categoryMod.Get(categoryParam)
+			if err != nil {
+				log.Warn("Failed to fetch categories: %v", err)
+			} else {
+				for _, category := range categories {
+					if categoryID, ok := category["category_id"].(string); ok {
+						if categoryName, ok := category["name"].(string); ok {
+							categoryMap[categoryID] = categoryName
+						}
+					}
+				}
+			}
+		}
+	}
+
+	// Add category_name to jobs
+	for i, job := range jobsData {
+		if categoryID, exists := job["category_id"]; exists && categoryID != nil {
+			if categoryIDStr, ok := categoryID.(string); ok {
+				if categoryName, exists := categoryMap[categoryIDStr]; exists {
+					jobsData[i]["category_name"] = categoryName
+				} else {
+					jobsData[i]["category_name"] = nil
+				}
+			}
+		}
+	}
+
+	result["data"] = jobsData
+	log.Debug("ListJobs result with categories: %+v", result)
+	return result, nil
 }
 
 // GetActiveJobs get active jobs (running, ready status)
@@ -33,6 +170,7 @@ func GetActiveJobs() ([]*Job, error) {
 	}
 
 	param := model.QueryParam{
+		Select: JobFields,
 		Wheres: []model.QueryWhere{
 			{Column: "status", OP: "in", Value: []string{"ready", "running"}},
 			{Column: "enabled", Value: true},
@@ -185,6 +323,7 @@ func GetJob(jobID string) (*Job, error) {
 	}
 
 	param := model.QueryParam{
+		Select: JobFields,
 		Wheres: []model.QueryWhere{
 			{Column: "job_id", Value: jobID},
 		},
@@ -199,8 +338,33 @@ func GetJob(jobID string) (*Job, error) {
 		return nil, fmt.Errorf("job not found: %s", jobID)
 	}
 
+	jobData := results[0]
+
+	// Query category name if category_id exists
+	if categoryID, exists := jobData["category_id"]; exists && categoryID != nil {
+		if categoryIDStr, ok := categoryID.(string); ok && categoryIDStr != "" {
+			categoryMod := model.Select("__yao.job.category")
+			if categoryMod != nil {
+				categoryParam := model.QueryParam{
+					Select: []interface{}{"name"},
+					Wheres: []model.QueryWhere{
+						{Column: "category_id", Value: categoryIDStr},
+					},
+					Limit: 1,
+				}
+
+				categoryResults, err := categoryMod.Get(categoryParam)
+				if err == nil && len(categoryResults) > 0 {
+					if categoryName, ok := categoryResults[0]["name"].(string); ok {
+						jobData["category_name"] = categoryName
+					}
+				}
+			}
+		}
+	}
+
 	job := &Job{}
-	if err := mapToStruct(results[0], job); err != nil {
+	if err := mapToStruct(jobData, job); err != nil {
 		return nil, err
 	}
 
@@ -218,6 +382,11 @@ func GetCategories(param model.QueryParam) ([]*Category, error) {
 		return nil, fmt.Errorf("job category model not found")
 	}
 
+	// Set select fields if not already specified
+	if len(param.Select) == 0 {
+		param.Select = CategoryFields
+	}
+
 	results, err := mod.Get(param)
 	if err != nil {
 		return nil, err
@@ -523,6 +692,11 @@ func ListLogs(jobID string, param model.QueryParam, page int, pagesize int) (map
 		return nil, fmt.Errorf("job log model not found")
 	}
 
+	// Set select fields if not already specified
+	if len(param.Select) == 0 {
+		param.Select = LogFields
+	}
+
 	// Add job_id filter
 	param.Wheres = append(param.Wheres, model.QueryWhere{
 		Column: "job_id",
@@ -597,6 +771,7 @@ func GetExecutions(jobID string) ([]*Execution, error) {
 	}
 
 	param := model.QueryParam{
+		Select: ExecutionFields,
 		Wheres: []model.QueryWhere{
 			{Column: "job_id", Value: jobID},
 		},
@@ -701,6 +876,11 @@ func GetExecution(executionID string, param model.QueryParam) (*Execution, error
 		return nil, fmt.Errorf("job execution model not found")
 	}
 
+	// Set select fields if not already specified
+	if len(param.Select) == 0 {
+		param.Select = ExecutionFields
+	}
+
 	param.Wheres = append(param.Wheres, model.QueryWhere{
 		Column: "execution_id",
 		Value:  executionID,

job/job.go

@@ -320,6 +320,14 @@ func makeJob(data []byte) (*Job, error) {
 		job.CreatedBy = "system"
 	}
 
+	// Set default enabled to true if not specified
+	// Note: Go's zero value for bool is false, so we need to explicitly check if it was set
+	// Since we can't distinguish between explicitly set false and zero value,
+	// we'll assume new jobs should be enabled by default
+	if !job.System && !job.Readonly {
+		job.Enabled = true
+	}
+
 	return &job, nil
 }
 

openapi/job/categories.go

@@ -0,0 +1,180 @@
+package job
+
+import (
+	"net/http"
+
+	"github.com/gin-gonic/gin"
+	"github.com/yaoapp/gou/model"
+	"github.com/yaoapp/gou/process"
+	"github.com/yaoapp/kun/log"
+	"github.com/yaoapp/yao/job"
+)
+
+// ListCategories lists job categories
+func ListCategories(c *gin.Context) {
+	// Build query parameters
+	param := model.QueryParam{}
+
+	// Add enabled filter (default to true)
+	enabled := c.DefaultQuery("enabled", "true")
+	if enabled == "true" {
+		param.Wheres = append(param.Wheres, model.QueryWhere{
+			Column: "enabled",
+			Value:  true,
+		})
+	}
+
+	// Add system filter if provided
+	if system := c.Query("system"); system != "" {
+		systemBool := system == "true"
+		param.Wheres = append(param.Wheres, model.QueryWhere{
+			Column: "system",
+			Value:  systemBool,
+		})
+	}
+
+	// Order by sort and name
+	param.Orders = []model.QueryOrder{
+		{Column: "sort", Option: "asc"},
+		{Column: "name", Option: "asc"},
+	}
+
+	// Get categories
+	categories, err := job.GetCategories(param)
+	if err != nil {
+		log.Error("Failed to list categories: %v", err)
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+
+	response := gin.H{
+		"data":  categories,
+		"total": len(categories),
+	}
+
+	c.JSON(http.StatusOK, response)
+}
+
+// GetCategory gets a specific category by ID
+func GetCategory(c *gin.Context) {
+	categoryID := c.Param("categoryID")
+	if categoryID == "" {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "category_id is required"})
+		return
+	}
+
+	// Build query parameters to find by category_id
+	param := model.QueryParam{
+		Wheres: []model.QueryWhere{
+			{Column: "category_id", Value: categoryID},
+		},
+		Limit: 1,
+	}
+
+	// Get categories with filter
+	categories, err := job.GetCategories(param)
+	if err != nil {
+		log.Error("Failed to get category %s: %v", categoryID, err)
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+
+	if len(categories) == 0 {
+		c.JSON(http.StatusNotFound, gin.H{"error": "Category not found"})
+		return
+	}
+
+	c.JSON(http.StatusOK, categories[0])
+}
+
+// ========================
+// Process Handlers
+// ========================
+
+// ProcessListCategories process handler for listing categories
+func ProcessListCategories(process *process.Process) interface{} {
+	// TODO: Implement process handler for listing categories
+	args := process.Args
+	log.Info("ProcessListCategories called with args: %v", args)
+
+	// Build query parameters
+	param := model.QueryParam{}
+	if len(args) > 0 {
+		if queryParam, ok := args[0].(model.QueryParam); ok {
+			param = queryParam
+		}
+	}
+
+	// Call job.GetCategories function
+	categories, err := job.GetCategories(param)
+	if err != nil {
+		log.Error("Failed to list categories: %v", err)
+		return map[string]interface{}{"error": err.Error()}
+	}
+
+	return map[string]interface{}{
+		"categories": categories,
+		"count":      len(categories),
+	}
+}
+
+// ProcessGetCategory process handler for getting a category
+func ProcessGetCategory(process *process.Process) interface{} {
+	// TODO: Implement process handler for getting a category
+	args := process.Args
+	if len(args) == 0 {
+		return map[string]interface{}{"error": "category_id is required"}
+	}
+
+	categoryID, ok := args[0].(string)
+	if !ok {
+		return map[string]interface{}{"error": "category_id must be a string"}
+	}
+
+	log.Info("ProcessGetCategory called for category: %s", categoryID)
+
+	// Build query parameters to find by category_id
+	param := model.QueryParam{
+		Wheres: []model.QueryWhere{
+			{Column: "category_id", Value: categoryID},
+		},
+		Limit: 1,
+	}
+
+	// Call job.GetCategories function with filter
+	categories, err := job.GetCategories(param)
+	if err != nil {
+		log.Error("Failed to get category %s: %v", categoryID, err)
+		return map[string]interface{}{"error": err.Error()}
+	}
+
+	if len(categories) == 0 {
+		return map[string]interface{}{"error": "category not found"}
+	}
+
+	return categories[0]
+}
+
+// ProcessCountCategories process handler for counting categories
+func ProcessCountCategories(process *process.Process) interface{} {
+	// TODO: Implement process handler for counting categories
+	args := process.Args
+	log.Info("ProcessCountCategories called with args: %v", args)
+
+	// Build query parameters
+	param := model.QueryParam{}
+	if len(args) > 0 {
+		if queryParam, ok := args[0].(model.QueryParam); ok {
+			param = queryParam
+		}
+	}
+
+	// Call job.CountCategories function
+	count, err := job.CountCategories(param)
+	if err != nil {
+		log.Error("Failed to count categories: %v", err)
+		return map[string]interface{}{"error": err.Error()}
+	}
+
+	return map[string]interface{}{"count": count}
+}

openapi/job/executions.go

@@ -0,0 +1,333 @@
+package job
+
+import (
+	"net/http"
+	"strconv"
+
+	"github.com/gin-gonic/gin"
+	"github.com/yaoapp/gou/model"
+	"github.com/yaoapp/gou/process"
+	"github.com/yaoapp/kun/log"
+	"github.com/yaoapp/yao/job"
+)
+
+// ListExecutions lists executions for a specific job
+func ListExecutions(c *gin.Context) {
+	jobID := c.Param("jobID")
+	if jobID == "" {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "job_id is required"})
+		return
+	}
+
+	// Get executions for the job
+	executions, err := job.GetExecutions(jobID)
+	if err != nil {
+		log.Error("Failed to list executions for job %s: %v", jobID, err)
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+
+	// Add optional status filter
+	if status := c.Query("status"); status != "" {
+		filtered := make([]*job.Execution, 0)
+		for _, execution := range executions {
+			if execution.Status == status {
+				filtered = append(filtered, execution)
+			}
+		}
+		executions = filtered
+	}
+
+	// Simple pagination (client-side)
+	page := 1
+	pagesize := 50
+	if p := c.Query("page"); p != "" {
+		if parsed, err := strconv.Atoi(p); err == nil && parsed > 0 {
+			page = parsed
+		}
+	}
+	if ps := c.Query("pagesize"); ps != "" {
+		if parsed, err := strconv.Atoi(ps); err == nil && parsed > 0 && parsed <= 1000 {
+			pagesize = parsed
+		}
+	}
+
+	total := len(executions)
+	start := (page - 1) * pagesize
+	end := start + pagesize
+
+	if start >= total {
+		executions = []*job.Execution{}
+	} else {
+		if end > total {
+			end = total
+		}
+		executions = executions[start:end]
+	}
+
+	response := gin.H{
+		"data":     executions,
+		"page":     page,
+		"pagesize": pagesize,
+		"total":    total,
+		"job_id":   jobID,
+	}
+
+	c.JSON(http.StatusOK, response)
+}
+
+// GetExecution gets a specific execution by ID
+func GetExecution(c *gin.Context) {
+	executionID := c.Param("executionID")
+	if executionID == "" {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "execution_id is required"})
+		return
+	}
+
+	// Get the execution
+	execution, err := job.GetExecution(executionID, model.QueryParam{})
+	if err != nil {
+		log.Error("Failed to get execution %s: %v", executionID, err)
+		if err.Error() == "execution not found: "+executionID {
+			c.JSON(http.StatusNotFound, gin.H{"error": "Execution not found"})
+		} else {
+			c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		}
+		return
+	}
+
+	c.JSON(http.StatusOK, execution)
+}
+
+// StopExecution stops a running execution
+func StopExecution(c *gin.Context) {
+	executionID := c.Param("executionID")
+	if executionID == "" {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "execution_id is required"})
+		return
+	}
+
+	// Get the execution first to find the job
+	execution, err := job.GetExecution(executionID, model.QueryParam{})
+	if err != nil {
+		log.Error("Failed to get execution %s: %v", executionID, err)
+		if err.Error() == "execution not found: "+executionID {
+			c.JSON(http.StatusNotFound, gin.H{"error": "Execution not found"})
+		} else {
+			c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		}
+		return
+	}
+
+	// Get the job to stop the specific execution
+	jobInstance, err := job.GetJob(execution.JobID)
+	if err != nil {
+		log.Error("Failed to get job %s for execution %s: %v", execution.JobID, executionID, err)
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+
+	// For now, we stop the entire job since individual execution stopping
+	// would require more complex implementation in the job package
+	err = jobInstance.Stop()
+	if err != nil {
+		log.Error("Failed to stop job %s (execution %s): %v", execution.JobID, executionID, err)
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+
+	c.JSON(http.StatusOK, gin.H{
+		"message":      "Execution stopped successfully (job stopped)",
+		"execution_id": executionID,
+		"job_id":       execution.JobID,
+		"status":       "stopped",
+	})
+}
+
+// GetExecutionProgress gets execution progress information
+func GetExecutionProgress(c *gin.Context) {
+	executionID := c.Param("executionID")
+	if executionID == "" {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "execution_id is required"})
+		return
+	}
+
+	// Get the execution
+	execution, err := job.GetExecution(executionID, model.QueryParam{})
+	if err != nil {
+		log.Error("Failed to get execution %s: %v", executionID, err)
+		if err.Error() == "execution not found: "+executionID {
+			c.JSON(http.StatusNotFound, gin.H{"error": "Execution not found"})
+		} else {
+			c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		}
+		return
+	}
+
+	response := gin.H{
+		"execution_id":  executionID,
+		"job_id":        execution.JobID,
+		"status":        execution.Status,
+		"progress":      execution.Progress,
+		"started_at":    execution.StartedAt,
+		"ended_at":      execution.EndedAt,
+		"duration":      execution.Duration,
+		"worker_id":     execution.WorkerID,
+		"process_id":    execution.ProcessID,
+		"retry_attempt": execution.RetryAttempt,
+	}
+
+	// Add error info if available
+	if execution.ErrorInfo != nil {
+		response["error_info"] = execution.ErrorInfo
+	}
+
+	// Add result if available
+	if execution.Result != nil {
+		response["result"] = execution.Result
+	}
+
+	c.JSON(http.StatusOK, response)
+}
+
+// ========================
+// Process Handlers
+// ========================
+
+// ProcessListExecutions process handler for listing executions
+func ProcessListExecutions(process *process.Process) interface{} {
+	// TODO: Implement process handler for listing executions
+	args := process.Args
+	if len(args) == 0 {
+		return map[string]interface{}{"error": "job_id is required"}
+	}
+
+	jobID, ok := args[0].(string)
+	if !ok {
+		return map[string]interface{}{"error": "job_id must be a string"}
+	}
+
+	log.Info("ProcessListExecutions called for job: %s", jobID)
+
+	// Call job.GetExecutions function
+	executions, err := job.GetExecutions(jobID)
+	if err != nil {
+		log.Error("Failed to list executions for job %s: %v", jobID, err)
+		return map[string]interface{}{"error": err.Error()}
+	}
+
+	return map[string]interface{}{
+		"executions": executions,
+		"count":      len(executions),
+	}
+}
+
+// ProcessGetExecution process handler for getting an execution
+func ProcessGetExecution(process *process.Process) interface{} {
+	// TODO: Implement process handler for getting an execution
+	args := process.Args
+	if len(args) == 0 {
+		return map[string]interface{}{"error": "execution_id is required"}
+	}
+
+	executionID, ok := args[0].(string)
+	if !ok {
+		return map[string]interface{}{"error": "execution_id must be a string"}
+	}
+
+	log.Info("ProcessGetExecution called for execution: %s", executionID)
+
+	// Build query parameters
+	param := model.QueryParam{}
+	if len(args) > 1 {
+		if queryParam, ok := args[1].(model.QueryParam); ok {
+			param = queryParam
+		}
+	}
+
+	// Call job.GetExecution function
+	execution, err := job.GetExecution(executionID, param)
+	if err != nil {
+		log.Error("Failed to get execution %s: %v", executionID, err)
+		return map[string]interface{}{"error": err.Error()}
+	}
+
+	return execution
+}
+
+// ProcessCountExecutions process handler for counting executions
+func ProcessCountExecutions(process *process.Process) interface{} {
+	// TODO: Implement process handler for counting executions
+	args := process.Args
+	if len(args) == 0 {
+		return map[string]interface{}{"error": "job_id is required"}
+	}
+
+	jobID, ok := args[0].(string)
+	if !ok {
+		return map[string]interface{}{"error": "job_id must be a string"}
+	}
+
+	log.Info("ProcessCountExecutions called for job: %s", jobID)
+
+	// Build query parameters
+	param := model.QueryParam{}
+	if len(args) > 1 {
+		if queryParam, ok := args[1].(model.QueryParam); ok {
+			param = queryParam
+		}
+	}
+
+	// Call job.CountExecutions function
+	count, err := job.CountExecutions(jobID, param)
+	if err != nil {
+		log.Error("Failed to count executions for job %s: %v", jobID, err)
+		return map[string]interface{}{"error": err.Error()}
+	}
+
+	return map[string]interface{}{"count": count}
+}
+
+// ProcessStopExecution process handler for stopping an execution
+func ProcessStopExecution(process *process.Process) interface{} {
+	// TODO: Implement process handler for stopping an execution
+	args := process.Args
+	if len(args) == 0 {
+		return map[string]interface{}{"error": "execution_id is required"}
+	}
+
+	executionID, ok := args[0].(string)
+	if !ok {
+		return map[string]interface{}{"error": "execution_id must be a string"}
+	}
+
+	log.Info("ProcessStopExecution called for execution: %s", executionID)
+
+	// Get the execution first to find the job
+	execution, err := job.GetExecution(executionID, model.QueryParam{})
+	if err != nil {
+		log.Error("Failed to get execution %s: %v", executionID, err)
+		return map[string]interface{}{"error": err.Error()}
+	}
+
+	// Get the job to stop the specific execution
+	jobInstance, err := job.GetJob(execution.JobID)
+	if err != nil {
+		log.Error("Failed to get job %s for execution %s: %v", execution.JobID, executionID, err)
+		return map[string]interface{}{"error": err.Error()}
+	}
+
+	// For now, we stop the entire job since individual execution stopping
+	// would require more complex implementation in the job package
+	err = jobInstance.Stop()
+	if err != nil {
+		log.Error("Failed to stop job %s (execution %s): %v", execution.JobID, executionID, err)
+		return map[string]interface{}{"error": err.Error()}
+	}
+
+	return map[string]interface{}{
+		"message":      "Execution stopped successfully",
+		"execution_id": executionID,
+		"job_id":       execution.JobID,
+	}
+}

openapi/job/job.go

@@ -1 +1,56 @@
 package job
+
+import (
+	"github.com/gin-gonic/gin"
+	"github.com/yaoapp/gou/process"
+	"github.com/yaoapp/yao/openapi/oauth/types"
+)
+
+func init() {
+	// Register job process handlers
+	process.RegisterGroup("job", map[string]process.Handler{
+		"jobs.list":        ProcessListJobs,
+		"jobs.get":         ProcessGetJob,
+		"jobs.count":       ProcessCountJobs,
+		"jobs.stop":        ProcessStopJob,
+		"executions.list":  ProcessListExecutions,
+		"executions.get":   ProcessGetExecution,
+		"executions.count": ProcessCountExecutions,
+		"executions.stop":  ProcessStopExecution,
+		"logs.list":        ProcessListLogs,
+		"categories.list":  ProcessListCategories,
+		"categories.get":   ProcessGetCategory,
+		"categories.count": ProcessCountCategories,
+	})
+}
+
+// Attach attaches the Job API to the router
+func Attach(group *gin.RouterGroup, oauth types.OAuth) {
+	// Protect all endpoints with OAuth
+	group.Use(oauth.Guard)
+
+	// Job Management (Read-only operations)
+	group.GET("/jobs", ListJobs)
+	group.GET("/jobs/:jobID", GetJob)
+	group.POST("/jobs/:jobID/stop", StopJob)
+
+	// Execution Management
+	group.GET("/jobs/:jobID/executions", ListExecutions)
+	group.GET("/executions/:executionID", GetExecution)
+	group.POST("/executions/:executionID/stop", StopExecution)
+
+	// Log Management
+	group.GET("/jobs/:jobID/logs", ListLogs)
+	group.GET("/executions/:executionID/logs", ListExecutionLogs)
+
+	// Category Management (Read-only)
+	group.GET("/categories", ListCategories)
+	group.GET("/categories/:categoryID", GetCategory)
+
+	// Progress and Status
+	group.GET("/jobs/:jobID/progress", GetJobProgress)
+	group.GET("/executions/:executionID/progress", GetExecutionProgress)
+
+	// Statistics
+	group.GET("/stats", GetStats)
+}

openapi/job/jobs.go

@@ -0,0 +1,404 @@
+package job
+
+import (
+	"net/http"
+	"strconv"
+
+	"github.com/gin-gonic/gin"
+	"github.com/yaoapp/gou/model"
+	"github.com/yaoapp/gou/process"
+	"github.com/yaoapp/kun/log"
+	"github.com/yaoapp/yao/job"
+)
+
+// ListJobs lists jobs with pagination
+func ListJobs(c *gin.Context) {
+	// Parse pagination parameters
+	page := 1
+	pagesize := 20
+
+	if p := c.Query("page"); p != "" {
+		if parsed, err := strconv.Atoi(p); err == nil && parsed > 0 {
+			page = parsed
+		}
+	}
+
+	if ps := c.Query("pagesize"); ps != "" {
+		if parsed, err := strconv.Atoi(ps); err == nil && parsed > 0 && parsed <= 1000 {
+			pagesize = parsed
+		}
+	}
+
+	// Build query parameters from URL query
+	param := model.QueryParam{
+		Orders: []model.QueryOrder{
+			{Column: "created_at", Option: "desc"},
+		},
+	}
+
+	// Add status filter if provided
+	if status := c.Query("status"); status != "" {
+		param.Wheres = append(param.Wheres, model.QueryWhere{
+			Column: "status",
+			Value:  status,
+		})
+	}
+
+	// Add category filter if provided
+	if categoryID := c.Query("category_id"); categoryID != "" {
+		param.Wheres = append(param.Wheres, model.QueryWhere{
+			Column: "category_id",
+			Value:  categoryID,
+		})
+	}
+
+	// Add enabled filter (default to show all for debugging)
+	enabled := c.Query("enabled")
+	if enabled == "true" {
+		param.Wheres = append(param.Wheres, model.QueryWhere{
+			Column: "enabled",
+			Value:  true,
+		})
+	} else if enabled == "false" {
+		param.Wheres = append(param.Wheres, model.QueryWhere{
+			Column: "enabled",
+			Value:  false,
+		})
+	}
+	// Default: show all records regardless of enabled status
+
+	// Call job.ListJobs function
+	result, err := job.ListJobs(param, page, pagesize)
+	if err != nil {
+		log.Error("Failed to list jobs: %v", err)
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+
+	c.JSON(http.StatusOK, result)
+}
+
+// GetJob gets a specific job by ID
+func GetJob(c *gin.Context) {
+	jobID := c.Param("jobID")
+	if jobID == "" {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "job_id is required"})
+		return
+	}
+
+	// Call job.GetJob function
+	jobInstance, err := job.GetJob(jobID)
+	if err != nil {
+		log.Error("Failed to get job %s: %v", jobID, err)
+		if err.Error() == "job not found: "+jobID {
+			c.JSON(http.StatusNotFound, gin.H{"error": "Job not found"})
+		} else {
+			c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		}
+		return
+	}
+
+	c.JSON(http.StatusOK, jobInstance)
+}
+
+// StopJob stops a running job
+func StopJob(c *gin.Context) {
+	jobID := c.Param("jobID")
+	if jobID == "" {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "job_id is required"})
+		return
+	}
+
+	// Get the job first
+	jobInstance, err := job.GetJob(jobID)
+	if err != nil {
+		log.Error("Failed to get job %s: %v", jobID, err)
+		if err.Error() == "job not found: "+jobID {
+			c.JSON(http.StatusNotFound, gin.H{"error": "Job not found"})
+		} else {
+			c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		}
+		return
+	}
+
+	// Stop the job
+	err = jobInstance.Stop()
+	if err != nil {
+		log.Error("Failed to stop job %s: %v", jobID, err)
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+
+	c.JSON(http.StatusOK, gin.H{
+		"message": "Job stopped successfully",
+		"job_id":  jobID,
+		"status":  "stopped",
+	})
+}
+
+// GetJobProgress gets job progress information
+func GetJobProgress(c *gin.Context) {
+	jobID := c.Param("jobID")
+	if jobID == "" {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "job_id is required"})
+		return
+	}
+
+	// Get the job first
+	jobInstance, err := job.GetJob(jobID)
+	if err != nil {
+		log.Error("Failed to get job %s: %v", jobID, err)
+		if err.Error() == "job not found: "+jobID {
+			c.JSON(http.StatusNotFound, gin.H{"error": "Job not found"})
+		} else {
+			c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		}
+		return
+	}
+
+	// Get executions for progress calculation
+	executions, err := job.GetExecutions(jobID)
+	if err != nil {
+		log.Error("Failed to get executions for job %s: %v", jobID, err)
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+
+	// Calculate progress
+	totalExecutions := len(executions)
+	completedCount := 0
+	runningCount := 0
+	failedCount := 0
+	totalProgress := 0
+
+	for _, execution := range executions {
+		totalProgress += execution.Progress
+		switch execution.Status {
+		case "completed":
+			completedCount++
+		case "running":
+			runningCount++
+		case "failed":
+			failedCount++
+		}
+	}
+
+	averageProgress := 0
+	if totalExecutions > 0 {
+		averageProgress = totalProgress / totalExecutions
+	}
+
+	response := gin.H{
+		"job_id":           jobID,
+		"status":           jobInstance.Status,
+		"progress":         averageProgress,
+		"total_executions": totalExecutions,
+		"completed_count":  completedCount,
+		"running_count":    runningCount,
+		"failed_count":     failedCount,
+		"last_run_at":      jobInstance.LastRunAt,
+		"next_run_at":      jobInstance.NextRunAt,
+	}
+
+	c.JSON(http.StatusOK, response)
+}
+
+// GetStats gets overall job statistics
+func GetStats(c *gin.Context) {
+	// Count total jobs
+	totalJobs, err := job.CountJobs(model.QueryParam{})
+	if err != nil {
+		log.Error("Failed to count total jobs: %v", err)
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+
+	// Count running jobs
+	runningJobs, err := job.CountJobs(model.QueryParam{
+		Wheres: []model.QueryWhere{
+			{Column: "status", Value: "running"},
+		},
+	})
+	if err != nil {
+		log.Error("Failed to count running jobs: %v", err)
+		runningJobs = 0
+	}
+
+	// Count completed jobs
+	completedJobs, err := job.CountJobs(model.QueryParam{
+		Wheres: []model.QueryWhere{
+			{Column: "status", Value: "completed"},
+		},
+	})
+	if err != nil {
+		log.Error("Failed to count completed jobs: %v", err)
+		completedJobs = 0
+	}
+
+	// Count failed jobs
+	failedJobs, err := job.CountJobs(model.QueryParam{
+		Wheres: []model.QueryWhere{
+			{Column: "status", Value: "failed"},
+		},
+	})
+	if err != nil {
+		log.Error("Failed to count failed jobs: %v", err)
+		failedJobs = 0
+	}
+
+	// Get categories for category stats
+	categories, err := job.GetCategories(model.QueryParam{})
+	if err != nil {
+		log.Error("Failed to get categories: %v", err)
+		categories = []*job.Category{}
+	}
+
+	categoryStats := make(map[string]int)
+	for _, category := range categories {
+		count, err := job.CountJobs(model.QueryParam{
+			Wheres: []model.QueryWhere{
+				{Column: "category_id", Value: category.CategoryID},
+			},
+		})
+		if err != nil {
+			count = 0
+		}
+		categoryStats[category.Name] = count
+	}
+
+	response := gin.H{
+		"total_jobs":       totalJobs,
+		"running_jobs":     runningJobs,
+		"completed_jobs":   completedJobs,
+		"failed_jobs":      failedJobs,
+		"category_stats":   categoryStats,
+		"total_categories": len(categories),
+	}
+
+	c.JSON(http.StatusOK, response)
+}
+
+// ========================
+// Process Handlers
+// ========================
+
+// ProcessListJobs process handler for listing jobs
+func ProcessListJobs(process *process.Process) interface{} {
+	// TODO: Implement process handler for listing jobs
+	args := process.Args
+	log.Info("ProcessListJobs called with args: %v", args)
+
+	// Default pagination values
+	page := 1
+	pagesize := 20
+
+	// Parse arguments if provided
+	if len(args) > 0 {
+		if p, ok := args[0].(int); ok {
+			page = p
+		}
+	}
+	if len(args) > 1 {
+		if ps, ok := args[1].(int); ok {
+			pagesize = ps
+		}
+	}
+
+	// Build query parameters
+	param := model.QueryParam{}
+	if len(args) > 2 {
+		if queryParam, ok := args[2].(model.QueryParam); ok {
+			param = queryParam
+		}
+	}
+
+	// Call job.ListJobs function
+	result, err := job.ListJobs(param, page, pagesize)
+	if err != nil {
+		log.Error("Failed to list jobs: %v", err)
+		return map[string]interface{}{"error": err.Error()}
+	}
+
+	return result
+}
+
+// ProcessGetJob process handler for getting a job
+func ProcessGetJob(process *process.Process) interface{} {
+	// TODO: Implement process handler for getting a job
+	args := process.Args
+	if len(args) == 0 {
+		return map[string]interface{}{"error": "job_id is required"}
+	}
+
+	jobID, ok := args[0].(string)
+	if !ok {
+		return map[string]interface{}{"error": "job_id must be a string"}
+	}
+
+	log.Info("ProcessGetJob called for job: %s", jobID)
+
+	// Call job.GetJob function
+	result, err := job.GetJob(jobID)
+	if err != nil {
+		log.Error("Failed to get job %s: %v", jobID, err)
+		return map[string]interface{}{"error": err.Error()}
+	}
+
+	return result
+}
+
+// ProcessCountJobs process handler for counting jobs
+func ProcessCountJobs(process *process.Process) interface{} {
+	// TODO: Implement process handler for counting jobs
+	args := process.Args
+	log.Info("ProcessCountJobs called with args: %v", args)
+
+	// Build query parameters
+	param := model.QueryParam{}
+	if len(args) > 0 {
+		if queryParam, ok := args[0].(model.QueryParam); ok {
+			param = queryParam
+		}
+	}
+
+	// Call job.CountJobs function
+	count, err := job.CountJobs(param)
+	if err != nil {
+		log.Error("Failed to count jobs: %v", err)
+		return map[string]interface{}{"error": err.Error()}
+	}
+
+	return map[string]interface{}{"count": count}
+}
+
+// ProcessStopJob process handler for stopping a job
+func ProcessStopJob(process *process.Process) interface{} {
+	// TODO: Implement process handler for stopping a job
+	args := process.Args
+	if len(args) == 0 {
+		return map[string]interface{}{"error": "job_id is required"}
+	}
+
+	jobID, ok := args[0].(string)
+	if !ok {
+		return map[string]interface{}{"error": "job_id must be a string"}
+	}
+
+	log.Info("ProcessStopJob called for job: %s", jobID)
+
+	// Get the job first
+	jobInstance, err := job.GetJob(jobID)
+	if err != nil {
+		log.Error("Failed to get job %s: %v", jobID, err)
+		return map[string]interface{}{"error": err.Error()}
+	}
+
+	// Stop the job
+	err = jobInstance.Stop()
+	if err != nil {
+		log.Error("Failed to stop job %s: %v", jobID, err)
+		return map[string]interface{}{"error": err.Error()}
+	}
+
+	return map[string]interface{}{"message": "Job stopped successfully", "job_id": jobID}
+}

openapi/job/logs.go

@@ -0,0 +1,188 @@
+package job
+
+import (
+	"net/http"
+	"strconv"
+
+	"github.com/gin-gonic/gin"
+	"github.com/yaoapp/gou/model"
+	"github.com/yaoapp/gou/process"
+	"github.com/yaoapp/kun/log"
+	"github.com/yaoapp/yao/job"
+)
+
+// ListLogs lists logs for a specific job
+func ListLogs(c *gin.Context) {
+	jobID := c.Param("jobID")
+	if jobID == "" {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "job_id is required"})
+		return
+	}
+
+	// Parse pagination parameters
+	page := 1
+	pagesize := 50
+
+	if p := c.Query("page"); p != "" {
+		if parsed, err := strconv.Atoi(p); err == nil && parsed > 0 {
+			page = parsed
+		}
+	}
+
+	if ps := c.Query("pagesize"); ps != "" {
+		if parsed, err := strconv.Atoi(ps); err == nil && parsed > 0 && parsed <= 1000 {
+			pagesize = parsed
+		}
+	}
+
+	// Build query parameters
+	param := model.QueryParam{
+		Orders: []model.QueryOrder{
+			{Column: "timestamp", Option: "desc"}, // 日志按时间戳倒序
+		},
+	}
+
+	// Add level filter if provided
+	if level := c.Query("level"); level != "" {
+		param.Wheres = append(param.Wheres, model.QueryWhere{
+			Column: "level",
+			Value:  level,
+		})
+	}
+
+	// Add execution_id filter if provided
+	if executionID := c.Query("execution_id"); executionID != "" {
+		param.Wheres = append(param.Wheres, model.QueryWhere{
+			Column: "execution_id",
+			Value:  executionID,
+		})
+	}
+
+	// Call job.ListLogs function
+	result, err := job.ListLogs(jobID, param, page, pagesize)
+	if err != nil {
+		log.Error("Failed to list logs for job %s: %v", jobID, err)
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+
+	c.JSON(http.StatusOK, result)
+}
+
+// ListExecutionLogs lists logs for a specific execution
+func ListExecutionLogs(c *gin.Context) {
+	executionID := c.Param("executionID")
+	if executionID == "" {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "execution_id is required"})
+		return
+	}
+
+	// First get the execution to find the job_id
+	execution, err := job.GetExecution(executionID, model.QueryParam{})
+	if err != nil {
+		log.Error("Failed to get execution %s: %v", executionID, err)
+		if err.Error() == "execution not found: "+executionID {
+			c.JSON(http.StatusNotFound, gin.H{"error": "Execution not found"})
+		} else {
+			c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		}
+		return
+	}
+
+	// Parse pagination parameters
+	page := 1
+	pagesize := 50
+
+	if p := c.Query("page"); p != "" {
+		if parsed, err := strconv.Atoi(p); err == nil && parsed > 0 {
+			page = parsed
+		}
+	}
+
+	if ps := c.Query("pagesize"); ps != "" {
+		if parsed, err := strconv.Atoi(ps); err == nil && parsed > 0 && parsed <= 1000 {
+			pagesize = parsed
+		}
+	}
+
+	// Build query parameters with execution_id filter
+	param := model.QueryParam{
+		Wheres: []model.QueryWhere{
+			{Column: "execution_id", Value: executionID},
+		},
+		Orders: []model.QueryOrder{
+			{Column: "timestamp", Option: "desc"}, // 日志按时间戳倒序
+		},
+	}
+
+	// Add level filter if provided
+	if level := c.Query("level"); level != "" {
+		param.Wheres = append(param.Wheres, model.QueryWhere{
+			Column: "level",
+			Value:  level,
+		})
+	}
+
+	// Call job.ListLogs function with job_id from execution
+	result, err := job.ListLogs(execution.JobID, param, page, pagesize)
+	if err != nil {
+		log.Error("Failed to list logs for execution %s: %v", executionID, err)
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+
+	c.JSON(http.StatusOK, result)
+}
+
+// ========================
+// Process Handlers
+// ========================
+
+// ProcessListLogs process handler for listing logs
+func ProcessListLogs(process *process.Process) interface{} {
+	// TODO: Implement process handler for listing logs
+	args := process.Args
+	if len(args) == 0 {
+		return map[string]interface{}{"error": "job_id is required"}
+	}
+
+	jobID, ok := args[0].(string)
+	if !ok {
+		return map[string]interface{}{"error": "job_id must be a string"}
+	}
+
+	// Default pagination values
+	page := 1
+	pagesize := 50
+
+	// Parse arguments if provided
+	if len(args) > 1 {
+		if p, ok := args[1].(int); ok && p > 0 {
+			page = p
+		}
+	}
+	if len(args) > 2 {
+		if ps, ok := args[2].(int); ok && ps > 0 {
+			pagesize = ps
+		}
+	}
+
+	// Build query parameters
+	param := model.QueryParam{}
+	if len(args) > 3 {
+		if queryParam, ok := args[3].(model.QueryParam); ok {
+			param = queryParam
+		}
+	}
+
+	log.Info("ProcessListLogs called for job: %s (page: %d, pagesize: %d)", jobID, page, pagesize)
+
+	// Call job.ListLogs function
+	result, err := job.ListLogs(jobID, param, page, pagesize)
+	if err != nil {
+		log.Error("Failed to list logs for job %s: %v", jobID, err)
+		return map[string]interface{}{"error": err.Error()}
+	}
+
+	return result
+}

openapi/openapi.go

@@ -11,6 +11,7 @@ import (
 	"github.com/yaoapp/yao/openapi/dsl"
 	"github.com/yaoapp/yao/openapi/file"
 	"github.com/yaoapp/yao/openapi/hello"
+	"github.com/yaoapp/yao/openapi/job"
 	"github.com/yaoapp/yao/openapi/kb"
 	"github.com/yaoapp/yao/openapi/oauth"
 	"github.com/yaoapp/yao/openapi/oauth/types"
@@ -95,6 +96,9 @@ func (openapi *OpenAPI) Attach(router *gin.Engine) {
 	// Knowledge Base handlers
 	kb.Attach(group.Group("/kb"), openapi.OAuth)
 
+	// Job Management handlers
+	job.Attach(group.Group("/job"), openapi.OAuth)
+
 	// Chat handlers
 	chat.Attach(group.Group("/chat"), openapi.OAuth)
 

yao/models/job/job.mod.yao

@@ -59,6 +59,14 @@
       "nullable": false,
       "index": true
     },
+    {
+      "name": "max_worker_nums",
+      "type": "integer",
+      "label": "Max Worker Numbers",
+      "comment": "Maximum number of concurrent workers for this job",
+      "default": 1,
+      "nullable": false
+    },
     {
       "name": "status",
       "type": "enum",
@@ -218,9 +226,9 @@
       "comment": "Composite index for category and status queries"
     },
     {
-      "name": "idx_job_process_type_status",
-      "columns": ["process_type", "status"],
-      "comment": "Composite index for process type and status queries"
+      "name": "idx_job_mode_status",
+      "columns": ["mode", "status"],
+      "comment": "Composite index for mode and status queries"
     },
     {
       "name": "idx_job_schedule_type_next_run",