// Package tasks implements a durable task execution plugin for Base. // // It provides persistent task queues with DAG-based workflow scheduling, // priority-based work-stealing, retry support, and timeout management, // all backed by Base's embedded SQLite. // // The plugin uses two internal collections: // - _tasks: individual work items with state machine lifecycle // - _workflows: ordered task DAGs (pipeline or fan-out) package tasks import ( "encoding/json" "time" "github.com/hanzoai/base/tools/types" ) // TaskState represents the lifecycle state of a task. type TaskState string const ( TaskPending TaskState = "pending" TaskClaimed TaskState = "claimed" TaskRunning TaskState = "running" TaskCompleted TaskState = "completed" TaskFailed TaskState = "failed" TaskCancelled TaskState = "cancelled" TaskRetrying TaskState = "retrying" ) // validTransitions defines which state transitions are allowed. var validTransitions = map[TaskState]map[TaskState]bool{ TaskPending: {TaskClaimed: true, TaskRunning: true, TaskCancelled: true}, TaskClaimed: {TaskRunning: true, TaskCancelled: true}, TaskRunning: {TaskCompleted: true, TaskFailed: true, TaskCancelled: true}, TaskFailed: {TaskRetrying: true, TaskCancelled: true}, TaskRetrying: {TaskPending: true, TaskCancelled: true}, TaskCompleted: {}, TaskCancelled: {}, } // CanTransition reports whether moving from one state to another is allowed. func CanTransition(from, to TaskState) bool { targets, ok := validTransitions[from] if !ok { return false } return targets[to] } // TaskPriority controls scheduling order. Higher values are scheduled first. type TaskPriority int const ( PriorityLow TaskPriority = 0 PriorityNormal TaskPriority = 1 PriorityHigh TaskPriority = 2 PriorityUrgent TaskPriority = 3 ) // Task is a durable work item. type Task struct { ID string `json:"id"` OrgID string `json:"org_id,omitempty"` // IAM org — maps to Temporal namespace SpaceID string `json:"space_id"` Title string `json:"title"` Description string `json:"description,omitempty"` State TaskState `json:"state"` Priority TaskPriority `json:"priority"` AssignedTo string `json:"assigned_to,omitempty"` CreatedBy string `json:"created_by,omitempty"` WorkflowID string `json:"workflow_id,omitempty"` ParentTaskID string `json:"parent_task_id,omitempty"` DependsOn []string `json:"depends_on,omitempty"` Labels []string `json:"labels,omitempty"` Input map[string]any `json:"input,omitempty"` Output map[string]any `json:"output,omitempty"` Error string `json:"error,omitempty"` Progress int `json:"progress"` MaxRetries int `json:"max_retries"` RetryCount int `json:"retry_count"` Timeout time.Duration `json:"timeout,omitempty"` CreatedAt time.Time `json:"created_at"` UpdatedAt time.Time `json:"updated_at"` StartedAt *time.Time `json:"started_at,omitempty"` CompletedAt *time.Time `json:"completed_at,omitempty"` Metadata map[string]string `json:"metadata,omitempty"` } // Workflow chains tasks into a DAG. type Workflow struct { ID string `json:"id"` OrgID string `json:"org_id,omitempty"` // IAM org — maps to Temporal namespace SpaceID string `json:"space_id"` Name string `json:"name"` Description string `json:"description,omitempty"` State TaskState `json:"state"` Tasks []string `json:"tasks"` CreatedBy string `json:"created_by,omitempty"` CreatedAt time.Time `json:"created_at"` UpdatedAt time.Time `json:"updated_at"` CompletedAt *time.Time `json:"completed_at,omitempty"` Metadata map[string]string `json:"metadata,omitempty"` } // TaskFilters controls listing and searching. type TaskFilters struct { OrgID string `json:"org_id,omitempty"` SpaceID string `json:"space_id,omitempty"` State *TaskState `json:"state,omitempty"` AssignedTo *string `json:"assigned_to,omitempty"` Priority *TaskPriority `json:"priority,omitempty"` Labels []string `json:"labels,omitempty"` WorkflowID *string `json:"workflow_id,omitempty"` Limit int `json:"limit,omitempty"` Offset int `json:"offset,omitempty"` } // marshalJSON marshals v to types.JSONRaw, returning nil on error. func marshalJSON(v any) types.JSONRaw { if v == nil { return nil } b, err := json.Marshal(v) if err != nil { return nil } return types.JSONRaw(b) } // unmarshalJSONField decodes a record's raw JSON field into dest. func unmarshalJSONField(raw any, dest any) { if raw == nil { return } switch v := raw.(type) { case types.JSONRaw: if len(v) > 0 { _ = json.Unmarshal(v, dest) } case string: if v != "" { _ = json.Unmarshal([]byte(v), dest) } case []byte: if len(v) > 0 { _ = json.Unmarshal(v, dest) } } }