diff --git a/.env.example b/.env.example index 0ea96ca..038de8b 100644 --- a/.env.example +++ b/.env.example @@ -186,6 +186,7 @@ CACHE_REDIS_CONN_MAX_LIFETIME=15m # Supported modes: embedded, queue TASK_RUNNER_MODE=embedded TASK_RUNNER_SHUTDOWN_TIMEOUT=30s +TASK_RUNNER_MAX_EXECUTION_TIME=1h TASK_RUNNER_RECOVERY_INTERVAL=30s TASK_RUNNER_HEARTBEAT_INTERVAL=30s TASK_RUNNER_MAX_CONCURRENT_TASKS=5 diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 17b96e8..e1b83c5 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -117,7 +117,7 @@ Service ports are fixed (`3000/3100/3200/3300/4000`); container-to-container URL - **Modular monolith over microservices** — one database, one deploy unit set, hard module boundaries enforced by convention (service interfaces only, no shared repositories). Scale comes from splitting *binaries*, not codebases. - **Transactional outbox** for anything that must not be lost; plain pub/sub for anything that may be. -- **Opinionated workflow model** — batch is the default: every node takes an array in and returns an array out, and references resolve per item (`$node.field`), by position (`$node[2].field`) or across the whole set (`$node[*].field`). What the canvas deliberately lacks is control-flow machinery — no loop construct, no sub-workflows, no expression language. Flows stay simple, readable, and predictable for non-developer users; that constraint is a feature. +- **Opinionated workflow model** — batch is the default: every node takes an array in and returns an array out, and references resolve per item (`$node.field` or `$input.field` for whatever feeds the node), by position (`$node[2].field`) or across the whole set (`$node[*].field`). Branching is per item: a condition sends each item down the branch its own comparison chose, and an output with no items skips whatever hangs off it. What the canvas deliberately lacks is control-flow machinery — no loop construct, no sub-workflows, no expression language. Flows stay simple, readable, and predictable for non-developer users; that constraint is a feature. - **stdlib-first Go** — `log/slog`, `database/sql`, small focused packages in `go-packages` instead of frameworks. - **Runtime env for the UI** — one image per release, environment decided at container start. - **Scale-out infrastructure is opt-in** — Redis buys cross-instance coordination, which a fresh single-instance install does not need. Every Redis-backed subsystem ships an in-process provider and defaults to it; Compose only starts what is configured. diff --git a/apps/platform-api/internal/config/task_runner.go b/apps/platform-api/internal/config/task_runner.go index f07928f..d50b99c 100644 --- a/apps/platform-api/internal/config/task_runner.go +++ b/apps/platform-api/internal/config/task_runner.go @@ -107,6 +107,7 @@ type TaskRunnerOptions struct { Mode TaskRunnerMode `env:"MODE"` ShutdownTimeout time.Duration `env:"SHUTDOWN_TIMEOUT"` + MaxExecutionTime time.Duration `env:"MAX_EXECUTION_TIME"` RecoveryInterval time.Duration `env:"RECOVERY_INTERVAL"` HeartbeatInterval time.Duration `env:"HEARTBEAT_INTERVAL"` MaxConcurrentTasks int64 `env:"MAX_CONCURRENT_TASKS"` diff --git a/apps/platform-api/internal/nodeengine/domain/executors/executor.go b/apps/platform-api/internal/nodeengine/domain/executors/executor.go index ea6b93c..63f6dd7 100644 --- a/apps/platform-api/internal/nodeengine/domain/executors/executor.go +++ b/apps/platform-api/internal/nodeengine/domain/executors/executor.go @@ -9,6 +9,10 @@ type Executor struct { Disabled bool `json:"disabled,omitempty"` } +type BranchingExecutor interface { + ExecuteBranches(ctx context.Context, credentials map[string]any, data []map[string]any) ([]map[string]any, map[string][]int, error) +} + type ExecutorManager interface { GetID() string GetDisabled() bool diff --git a/apps/platform-api/internal/nodeengine/nodes/system/condition/branch.go b/apps/platform-api/internal/nodeengine/nodes/system/condition/branch.go new file mode 100644 index 0000000..1e76a5a --- /dev/null +++ b/apps/platform-api/internal/nodeengine/nodes/system/condition/branch.go @@ -0,0 +1,6 @@ +package condition + +const ( + BranchTrue = "true" + BranchFalse = "false" +) diff --git a/apps/platform-api/internal/nodeengine/nodes/system/condition/executor.go b/apps/platform-api/internal/nodeengine/nodes/system/condition/executor.go index 80accd9..f827f59 100644 --- a/apps/platform-api/internal/nodeengine/nodes/system/condition/executor.go +++ b/apps/platform-api/internal/nodeengine/nodes/system/condition/executor.go @@ -38,49 +38,70 @@ func NewConditionExecutor( } func (e *ConditionExecutor) ExecuteWithContext(ctx context.Context, credentials map[string]any, data []map[string]any) ([]map[string]any, error) { + outputs, _, err := e.ExecuteBranches(ctx, credentials, data) + return outputs, err +} + +func (e *ConditionExecutor) ExecuteBranches(ctx context.Context, credentials map[string]any, data []map[string]any) ([]map[string]any, map[string][]int, error) { select { case <-ctx.Done(): - return nil, ctx.Err() + return nil, nil, ctx.Err() default: - // TODO: only the first item is evaluated, so a node fed 10 items decides - // the whole flow from item 0. Routing items per branch needs the runner - // to carry item identity and keep outputs per edge; see the branching - // notes before changing this. - input, err := e.validator.Parse(data[0]) - if err != nil { - return nil, err + outputs := make([]map[string]any, 0, len(data)) + branches := map[string][]int{ + BranchTrue: make([]int, 0, len(data)), + BranchFalse: make([]int, 0, len(data)), } - result := false - switch input.Operator { - case "eq": - result = input.LeftValue == input.RightValue - case "neq": - result = input.LeftValue != input.RightValue - case "gt": - result = cast.ToFloat(input.LeftValue) > cast.ToFloat(input.RightValue) - case "gte": - result = cast.ToFloat(input.LeftValue) >= cast.ToFloat(input.RightValue) - case "lt": - result = cast.ToFloat(input.LeftValue) < cast.ToFloat(input.RightValue) - case "lte": - result = cast.ToFloat(input.LeftValue) <= cast.ToFloat(input.RightValue) - case "contains": - result = strings.Contains(input.LeftValue, input.RightValue) - case "not_contains": - result = !strings.Contains(input.LeftValue, input.RightValue) - case "is_empty": - result = input.LeftValue == "" - case "is_not_empty": - result = input.LeftValue != "" - default: - return nil, ErrInvalidOperator + for index, item := range data { + result, err := e.evaluate(item) + if err != nil { + return nil, nil, err + } + + outputs = append(outputs, map[string]any{"status": result}) + if result { + branches[BranchTrue] = append(branches[BranchTrue], index) + continue + } + branches[BranchFalse] = append(branches[BranchFalse], index) } - return []map[string]any{ - { - "status": result, - }, - }, nil + return outputs, branches, nil + } +} + +func (e *ConditionExecutor) evaluate(item map[string]any) (bool, error) { + input, err := e.validator.Parse(item) + if err != nil { + return false, err + } + + left := strings.TrimSpace(input.LeftValue) + right := strings.TrimSpace(input.RightValue) + + switch input.Operator { + case "eq": + return left == right, nil + case "neq": + return left != right, nil + case "gt": + return cast.ToFloat(left) > cast.ToFloat(right), nil + case "gte": + return cast.ToFloat(left) >= cast.ToFloat(right), nil + case "lt": + return cast.ToFloat(left) < cast.ToFloat(right), nil + case "lte": + return cast.ToFloat(left) <= cast.ToFloat(right), nil + case "contains": + return strings.Contains(left, right), nil + case "not_contains": + return !strings.Contains(left, right), nil + case "is_empty": + return left == "", nil + case "is_not_empty": + return left != "", nil + default: + return false, ErrInvalidOperator } } diff --git a/apps/platform-api/internal/nodeengine/nodes/system/condition/node.go b/apps/platform-api/internal/nodeengine/nodes/system/condition/node.go index a67f110..e351fca 100644 --- a/apps/platform-api/internal/nodeengine/nodes/system/condition/node.go +++ b/apps/platform-api/internal/nodeengine/nodes/system/condition/node.go @@ -25,8 +25,8 @@ func NewConditionNode(nodeID string) *ConditionNode { {Key: "in"}, }, Outputs: []nodes.NodeHandle{ - {Key: "true", Label: "True"}, - {Key: "false", Label: "False"}, + {Key: BranchTrue, Label: "True"}, + {Key: BranchFalse, Label: "False"}, }, Categories: []string{"System"}, SubCategories: []string{"System"}, diff --git a/apps/platform-api/internal/taskrunner/application/taskrunner/branching.go b/apps/platform-api/internal/taskrunner/application/taskrunner/branching.go new file mode 100644 index 0000000..18cdc8b --- /dev/null +++ b/apps/platform-api/internal/taskrunner/application/taskrunner/branching.go @@ -0,0 +1,186 @@ +package taskrunner + +import ( + "strings" + + "github.com/blocknextai/go-packages/dag" + nodeEngineDomainExecutors "github.com/blocknextai/platform-api/internal/nodeengine/domain/executors" + taskRunnerDomainTask "github.com/blocknextai/platform-api/internal/taskrunner/domain/task" + "github.com/google/uuid" +) + +const branchKeySeparator = "#" + +func BuildBranchKey(nodeKey, handle string) string { + var builder strings.Builder + builder.WriteString(nodeKey) + builder.WriteString(branchKeySeparator) + builder.WriteString(handle) + return builder.String() +} + +type BranchView struct { + TaskID uuid.UUID + Keys map[string]string + Absolute []int + InputKey string + collected map[string][]any + matches map[string][][]string +} + +func (v *BranchView) Matches(pattern, template string, find func(string) [][]string) [][]string { + if v == nil { + return find(template) + } + if v.matches == nil { + v.matches = make(map[string][][]string) + } + key := pattern + "\x00" + template + if cached, ok := v.matches[key]; ok { + return cached + } + found := find(template) + v.matches[key] = found + return found +} + +func (v *BranchView) Collected(key string) ([]any, bool) { + if v == nil || v.collected == nil { + return nil, false + } + values, ok := v.collected[key] + return values, ok +} + +func (v *BranchView) Collect(key string, values []any) { + if v == nil { + return + } + if v.collected == nil { + v.collected = make(map[string][]any) + } + v.collected[key] = values +} + +func (v *BranchView) Input() string { + if v == nil { + return "" + } + return v.InputKey +} + +func (v *BranchView) Task() uuid.UUID { + if v == nil { + return uuid.Nil + } + return v.TaskID +} + +func (v *BranchView) StoreKey(nodeKey string) string { + if v == nil { + return nodeKey + } + if key, ok := v.Keys[nodeKey]; ok { + return key + } + return nodeKey +} + +func (v *BranchView) AbsoluteIndex(itemIndex int) int { + if v == nil || v.Absolute == nil { + return itemIndex + } + if itemIndex < 0 || itemIndex >= len(v.Absolute) { + return itemIndex + } + return v.Absolute[itemIndex] +} + +func (v *BranchView) ItemIndex(nodeKey string, itemIndex int) int { + if v == nil { + return itemIndex + } + if _, isBranched := v.Keys[nodeKey]; isBranched { + return itemIndex + } + return v.AbsoluteIndex(itemIndex) +} + +func edgeBetween(d *dag.DAG, parentID, childID string) *dag.Edge { + for _, edge := range d.NodeEdges(parentID) { + if edge.Target == childID { + return &edge + } + } + return nil +} + +func (e *NodeExecutor) buildBranchView(task *taskRunnerDomainTask.Task, node *dag.Node) *BranchView { + if task.DAG == nil { + return nil + } + + view := &BranchView{TaskID: task.ID, Keys: make(map[string]string)} + var builder strings.Builder + + parents := task.DAG.NodeParents(node.ID) + for _, parentID := range parents { + parentNode := task.DAG.NodeByID(parentID) + if parentNode == nil { + continue + } + + parentKey := BuildNodeKeyWithBuilder(&builder, parentNode.NodeID, parentNode.ID) + storeKey := parentKey + + edge := edgeBetween(task.DAG, parentID, node.ID) + if edge != nil && strings.TrimSpace(edge.SourceHandle) != "" { + branchKey := BuildBranchKey(parentKey, edge.SourceHandle) + if _, ok := e.outputStore.Get(task.ID, branchKey); ok { + storeKey = branchKey + view.Keys[parentKey] = branchKey + if view.Absolute == nil { + if absolute, ok := e.outputStore.Indexes(task.ID, branchKey); ok { + view.Absolute = absolute + } + } + } + } + + if len(parents) == 1 { + view.InputKey = storeKey + } + } + + return view +} + +func branchReaches(store *OutputStore, taskID uuid.UUID, d *dag.DAG, parent *dag.Node, childID string) (bool, error) { + edge := edgeBetween(d, parent.ID, childID) + if edge == nil || strings.TrimSpace(edge.SourceHandle) == "" { + return true, nil + } + + executor, ok := nodeEngineDomainExecutors.GetExecutor(parent.NodeID) + if !ok { + return true, nil + } + if _, routes := executor.(nodeEngineDomainExecutors.BranchingExecutor); !routes { + return true, nil + } + + items, ok := store.Get(taskID, BuildBranchKey(BuildNodeKey(parent.NodeID, parent.ID), edge.SourceHandle)) + if !ok { + return false, ErrBranchOutputsUnavailable + } + return len(items) > 0, nil +} + +func routesItems(node *dag.Node) bool { + executor, ok := nodeEngineDomainExecutors.GetExecutor(node.NodeID) + if !ok { + return false + } + _, routes := executor.(nodeEngineDomainExecutors.BranchingExecutor) + return routes +} diff --git a/apps/platform-api/internal/taskrunner/application/taskrunner/data_processor.go b/apps/platform-api/internal/taskrunner/application/taskrunner/data_processor.go index 3673e94..a7d74b9 100644 --- a/apps/platform-api/internal/taskrunner/application/taskrunner/data_processor.go +++ b/apps/platform-api/internal/taskrunner/application/taskrunner/data_processor.go @@ -1,6 +1,7 @@ package taskrunner import ( + "log/slog" "regexp" "strconv" "strings" @@ -22,17 +23,18 @@ var ( methodCallRegex = regexp.MustCompile(`\$([\w.]+_[\w-]+)\.(first|last|get)\((\d*)\)\.([a-zA-Z0-9_.]+)`) arrayAccessRegex = regexp.MustCompile(`\$([\w.]+_[\w-]+)(?:\[(\d+|\*)\])?\.([a-zA-Z0-9_.]+)`) triggerVarRegex = regexp.MustCompile(`\$trigger\.([\w.]+)`) + inputRegex = regexp.MustCompile(`\$input\.([a-zA-Z0-9_.]+)`) ) -func (p *DataProcessor) ProcessNodeData(data map[string]any, itemIndex int, triggerData map[string]any) map[string]any { +func (p *DataProcessor) ProcessNodeData(data map[string]any, itemIndex int, triggerData map[string]any, view *BranchView) map[string]any { processed := make(map[string]any, len(data)) for k, v := range data { switch value := v.(type) { case string: - processed[k] = p.processStringValue(value, itemIndex, triggerData) + processed[k] = p.processStringValue(value, itemIndex, triggerData, view) case map[string]any: - processed[k] = p.ProcessNodeData(value, itemIndex, triggerData) + processed[k] = p.ProcessNodeData(value, itemIndex, triggerData, view) default: processed[k] = v } @@ -40,24 +42,50 @@ func (p *DataProcessor) ProcessNodeData(data map[string]any, itemIndex int, trig return processed } -func (p *DataProcessor) processStringValue(value string, currentItemIndex int, triggerData map[string]any) any { +func (p *DataProcessor) processStringValue(value string, currentItemIndex int, triggerData map[string]any, view *BranchView) any { resolvedStr := value if triggerData != nil { resolvedStr = p.resolveTriggerVariables(resolvedStr, triggerData) } - resolvedStr = p.resolveMethodCalls(resolvedStr) + resolvedStr = p.resolveInputReferences(resolvedStr, currentItemIndex, view) - resolvedStr = p.resolveArrayAccess(resolvedStr, currentItemIndex) + resolvedStr = p.resolveMethodCalls(resolvedStr, view) + + resolvedStr = p.resolveArrayAccess(resolvedStr, currentItemIndex, view) return resolvedStr } +func substitute(template string, matches [][]string, resolve func(match []string) (string, any)) string { + resolved := template + + for _, match := range matches { + placeholder, value := resolve(match) + if placeholder == "" || value == nil || isComplexType(value) { + continue + } + + replacement := cast.ToString(value) + if strings.TrimSpace(template) != placeholder { + var builder strings.Builder + builder.WriteString(`"`) + builder.WriteString(replacement) + builder.WriteString(`"`) + replacement = builder.String() + } + + resolved = strings.ReplaceAll(resolved, placeholder, replacement) + } + + return resolved +} + func (p *DataProcessor) resolveTriggerVariables(value string, triggerData map[string]any) string { return triggerVarRegex.ReplaceAllStringFunc(value, func(match string) string { path := match[len("$trigger."):] - resolved := p.getNestedValueWithArrayAccess(triggerData, path) + resolved := getNestedValueWithArrayAccess(triggerData, path) if resolved == nil { return match } @@ -65,95 +93,103 @@ func (p *DataProcessor) resolveTriggerVariables(value string, triggerData map[st }) } -func (p *DataProcessor) resolveMethodCalls(value string) string { +func (p *DataProcessor) resolveInputReferences(value string, currentItemIndex int, view *BranchView) string { + if p.outputStore == nil || view.Input() == "" { + return value + } + + matches := view.Matches("input", value, func(template string) [][]string { + return inputRegex.FindAllStringSubmatch(template, -1) + }) + if len(matches) == 0 { + return value + } + + inputOutputs, ok := p.outputStore.Get(view.Task(), view.Input()) + if !ok || len(inputOutputs) == 0 { + return value + } + + itemIndex := currentItemIndex + if itemIndex >= len(inputOutputs) { + itemIndex = 0 + } + + return substitute(value, matches, func(match []string) (string, any) { + if len(match) < 2 { + return "", nil + } + + fieldPath := match[1] + return "$input." + fieldPath, getNestedValueWithArrayAccess(inputOutputs[itemIndex], fieldPath) + }) +} + +func (p *DataProcessor) resolveMethodCalls(value string, view *BranchView) string { if p.outputStore == nil { return value } - matches := methodCallRegex.FindAllStringSubmatch(value, -1) - resolvedStr := value + matches := view.Matches("method", value, func(template string) [][]string { + return methodCallRegex.FindAllStringSubmatch(template, -1) + }) - for _, match := range matches { + return substitute(value, matches, func(match []string) (string, any) { if len(match) < 5 { - continue + return "", nil } - nodeKey := match[1] - method := match[2] - methodArg := match[3] - fieldPath := match[4] + nodeKey, method, methodArg, fieldPath := match[1], match[2], match[3], match[4] - nodeOutputs, ok := p.outputStore.Get(nodeKey) + nodeOutputs, ok := p.outputStore.Get(view.Task(), view.StoreKey(nodeKey)) if !ok || len(nodeOutputs) == 0 { - continue + return "", nil } - targetResult := p.getTargetResult(nodeOutputs, method, methodArg) + targetResult := getTargetResult(nodeOutputs, method, methodArg) if targetResult == nil { - continue + return "", nil } - resolvedValue := p.getNestedValueWithArrayAccess(targetResult, fieldPath) - if resolvedValue == nil || p.isComplexType(resolvedValue) { - continue - } - - placeholder := p.buildMethodCallPlaceholder(nodeKey, method, methodArg, fieldPath) - - var b strings.Builder - b.WriteString(`"`) - b.WriteString(cast.ToString(resolvedValue)) - b.WriteString(`"`) - replacementValue := b.String() - - resolvedStr = strings.ReplaceAll(resolvedStr, placeholder, replacementValue) - } - - return resolvedStr + return buildMethodCallPlaceholder(nodeKey, method, methodArg, fieldPath), + getNestedValueWithArrayAccess(targetResult, fieldPath) + }) } -func (p *DataProcessor) resolveArrayAccess(value string, currentItemIndex int) string { +func (p *DataProcessor) resolveArrayAccess(value string, currentItemIndex int, view *BranchView) string { if p.outputStore == nil { return value } - matches := arrayAccessRegex.FindAllStringSubmatch(value, -1) - resolvedStr := value + matches := view.Matches("array", value, func(template string) [][]string { + return arrayAccessRegex.FindAllStringSubmatch(template, -1) + }) - for _, match := range matches { + return substitute(value, matches, func(match []string) (string, any) { if len(match) < 4 { - continue + return "", nil } - nodeKey := match[1] - indexStr := match[2] - fieldPath := match[3] + nodeKey, indexStr, fieldPath := match[1], match[2], match[3] - nodeOutputs, ok := p.outputStore.Get(nodeKey) - if !ok || len(nodeOutputs) == 0 { - continue + nodeOutputs, ok := p.outputStore.Get(view.Task(), view.StoreKey(nodeKey)) + if !ok { + slog.Warn("reference has no stored outputs to resolve against", + "component", "data_processor", + "task_id", view.Task(), + "node_key", nodeKey) + return "", nil } - - resolvedValue := p.resolveByIndex(nodeOutputs, indexStr, fieldPath, currentItemIndex) - if resolvedValue == nil || p.isComplexType(resolvedValue) { - continue + if len(nodeOutputs) == 0 { + return "", nil } - placeholder := p.buildArrayAccessPlaceholder(nodeKey, indexStr, fieldPath) - - var b strings.Builder - b.WriteString(`"`) - b.WriteString(cast.ToString(resolvedValue)) - b.WriteString(`"`) - replacementValue := b.String() - - resolvedStr = strings.ReplaceAll(resolvedStr, placeholder, replacementValue) - } - - return resolvedStr + return buildArrayAccessPlaceholder(nodeKey, indexStr, fieldPath), + resolveByIndex(nodeOutputs, nodeKey, indexStr, fieldPath, view.ItemIndex(nodeKey, currentItemIndex), view) + }) } -func (p *DataProcessor) getTargetResult(nodeOutputs []map[string]any, method, methodArg string) map[string]any { +func getTargetResult(nodeOutputs []map[string]any, method, methodArg string) map[string]any { switch method { case "first": if len(nodeOutputs) > 0 { @@ -173,43 +209,49 @@ func (p *DataProcessor) getTargetResult(nodeOutputs []map[string]any, method, me return nil } -func (p *DataProcessor) resolveByIndex(nodeOutputs []map[string]any, indexStr, fieldPath string, currentItemIndex int) any { +func resolveByIndex( + nodeOutputs []map[string]any, + nodeKey, indexStr, fieldPath string, + currentItemIndex int, + view *BranchView, +) any { switch indexStr { case "": if currentItemIndex < len(nodeOutputs) { - return p.getNestedValueWithArrayAccess(nodeOutputs[currentItemIndex], fieldPath) + return getNestedValueWithArrayAccess(nodeOutputs[currentItemIndex], fieldPath) } if len(nodeOutputs) > 0 { - return p.getNestedValueWithArrayAccess(nodeOutputs[0], fieldPath) + return getNestedValueWithArrayAccess(nodeOutputs[0], fieldPath) } case "*": + cacheKey := nodeKey + "[*]." + fieldPath + if cached, ok := view.Collected(cacheKey); ok { + if len(cached) > 0 { + return cached + } + return nil + } + allValues := make([]any, 0, len(nodeOutputs)) for _, result := range nodeOutputs { - if val := p.getNestedValueWithArrayAccess(result, fieldPath); val != nil { + if val := getNestedValueWithArrayAccess(result, fieldPath); val != nil { allValues = append(allValues, val) } } + view.Collect(cacheKey, allValues) + if len(allValues) > 0 { return allValues } default: if index, err := strconv.Atoi(indexStr); err == nil && index >= 0 && index < len(nodeOutputs) { - return p.getNestedValueWithArrayAccess(nodeOutputs[index], fieldPath) + return getNestedValueWithArrayAccess(nodeOutputs[index], fieldPath) } } return nil } -func (p *DataProcessor) isComplexType(value any) bool { - switch value.(type) { - case map[string]any, []any, []map[string]any: - return true - default: - return false - } -} - -func (p *DataProcessor) buildMethodCallPlaceholder(nodeKey, method, methodArg, fieldPath string) string { +func buildMethodCallPlaceholder(nodeKey, method, methodArg, fieldPath string) string { var builder strings.Builder builder.WriteString("$") builder.WriteString(nodeKey) @@ -224,7 +266,7 @@ func (p *DataProcessor) buildMethodCallPlaceholder(nodeKey, method, methodArg, f return builder.String() } -func (p *DataProcessor) buildArrayAccessPlaceholder(nodeKey, indexStr, fieldPath string) string { +func buildArrayAccessPlaceholder(nodeKey, indexStr, fieldPath string) string { var builder strings.Builder builder.WriteString("$") builder.WriteString(nodeKey) @@ -237,111 +279,3 @@ func (p *DataProcessor) buildArrayAccessPlaceholder(nodeKey, indexStr, fieldPath builder.WriteString(fieldPath) return builder.String() } - -func (p *DataProcessor) getNestedValueWithArrayAccess(data any, path string) any { - parts := p.splitPathPreservingParentheses(path) - if len(parts) == 0 { - return data - } - - current := data - part := parts[0] - - if strings.Contains(part, "(") && strings.Contains(part, ")") { - current = p.getArrayAccessValue(current, part) - } else { - currentMap, ok := current.(map[string]any) - if !ok { - return nil - } - val, ok := currentMap[part] - if !ok { - return nil - } - current = val - } - - if len(parts) == 1 { - return p.getFirstElementIfSlice(current) - } - - return p.getNestedValueWithArrayAccess(current, strings.Join(parts[1:], ".")) -} - -func (p *DataProcessor) getFirstElementIfSlice(current any) any { - if arr, ok := current.([]any); ok && len(arr) > 0 { - return arr[0] - } - if arr, ok := current.([]map[string]any); ok && len(arr) > 0 { - return arr[0] - } - return current -} - -func (p *DataProcessor) splitPathPreservingParentheses(path string) []string { - var parts []string - var current strings.Builder - parenCount := 0 - - for _, char := range path { - switch char { - case '(': - parenCount++ - case ')': - parenCount-- - } - - if char == '.' && parenCount == 0 { - if current.Len() > 0 { - parts = append(parts, current.String()) - current.Reset() - } - } else { - current.WriteRune(char) - } - } - - if current.Len() > 0 { - parts = append(parts, current.String()) - } - - return parts -} - -func (p *DataProcessor) getArrayAccessValue(data any, accessor string) any { - methodStart := strings.Index(accessor, "(") - methodEnd := strings.Index(accessor, ")") - - if methodStart == -1 || methodEnd == -1 { - return nil - } - - methodName := accessor[:methodStart] - argsStr := accessor[methodStart+1 : methodEnd] - - slice := cast.ToSlice(data) - if len(slice) == 0 { - return nil - } - - switch methodName { - case "get": - index, err := strconv.Atoi(argsStr) - if err != nil || index < 0 || index >= len(slice) { - return nil - } - return slice[index] - case "first": - if len(slice) == 0 { - return nil - } - return slice[0] - case "last": - if len(slice) == 0 { - return nil - } - return slice[len(slice)-1] - default: - return nil - } -} diff --git a/apps/platform-api/internal/taskrunner/application/taskrunner/errors.go b/apps/platform-api/internal/taskrunner/application/taskrunner/errors.go index f7a170c..7fbafa9 100644 --- a/apps/platform-api/internal/taskrunner/application/taskrunner/errors.go +++ b/apps/platform-api/internal/taskrunner/application/taskrunner/errors.go @@ -5,20 +5,22 @@ import ( ) var ( - ErrNodeExecutionNotFound = apperror.NotFound("node execution not found") - ErrTaskAlreadyCompleted = apperror.Conflict("task already completed") - ErrNodeExecutionIDNotFound = apperror.NotFound("node execution id not found") - ErrExecutorNotFound = apperror.NotFound("executor not found") - ErrTaskNotFound = apperror.NotFound("task not found") - ErrTaskNotFailed = apperror.Conflict("task not failed") - ErrTaskDAGIsEmpty = apperror.Validation("task dag is empty") - ErrNoStartNodesFound = apperror.Validation("no start nodes found") - ErrNoNodesFoundInWorkflow = apperror.Validation("no nodes found in workflow") - ErrNoNodeExecutionsFound = apperror.NotFound("no node executions found") - ErrSomeNodesFailed = apperror.Internal("some nodes failed") - ErrNodeDisabled = apperror.Conflict("node disabled") - ErrTaskCancelled = apperror.Internal("task cancelled") - ErrTaskExecutionPanic = apperror.Internal("task execution panic") - ErrWorkerPoolFull = apperror.Unavailable("worker pool full") - ErrTaskClaimNotAvailable = apperror.Conflict("task claim not available") + ErrTaskExecutionTimeLimit = apperror.Internal("task exceeded its execution time limit") + ErrBranchOutputsUnavailable = apperror.Internal("branch outputs are no longer available") + ErrNodeExecutionNotFound = apperror.NotFound("node execution not found") + ErrTaskAlreadyCompleted = apperror.Conflict("task already completed") + ErrNodeExecutionIDNotFound = apperror.NotFound("node execution id not found") + ErrExecutorNotFound = apperror.NotFound("executor not found") + ErrTaskNotFound = apperror.NotFound("task not found") + ErrTaskNotFailed = apperror.Conflict("task not failed") + ErrTaskDAGIsEmpty = apperror.Validation("task dag is empty") + ErrNoStartNodesFound = apperror.Validation("no start nodes found") + ErrNoNodesFoundInWorkflow = apperror.Validation("no nodes found in workflow") + ErrNoNodeExecutionsFound = apperror.NotFound("no node executions found") + ErrSomeNodesFailed = apperror.Internal("some nodes failed") + ErrNodeDisabled = apperror.Conflict("node disabled") + ErrTaskCancelled = apperror.Internal("task cancelled") + ErrTaskExecutionPanic = apperror.Internal("task execution panic") + ErrWorkerPoolFull = apperror.Unavailable("worker pool full") + ErrTaskClaimNotAvailable = apperror.Conflict("task claim not available") ) diff --git a/apps/platform-api/internal/taskrunner/application/taskrunner/node_executor.go b/apps/platform-api/internal/taskrunner/application/taskrunner/node_executor.go index 8e9968b..71f50aa 100644 --- a/apps/platform-api/internal/taskrunner/application/taskrunner/node_executor.go +++ b/apps/platform-api/internal/taskrunner/application/taskrunner/node_executor.go @@ -75,13 +75,14 @@ func (e *NodeExecutor) ExecuteNode(ctx context.Context, task *taskRunnerDomainTa } } - processedDataList := e.prepareAndProcessNodeData(task, node) + view := e.buildBranchView(task, node) + processedDataList := e.prepareAndProcessNodeData(task, node, view) e.updateNodeExecution(ctx, task.ID, task.NodeExecutionIDMap[node.ID], taskRunnerDomain.StatusRunning, processedDataList, nil, nil, nil, &startTime, nil) return e.executeWithRetries(ctx, task, node, executor, processedCredentials, - processedDataList, node.Parameters, maxRetries, retryDelay, timeout, startTime) + processedDataList, node.Parameters, maxRetries, retryDelay, timeout, startTime, view) } func (e *NodeExecutor) executeWithRetries( @@ -94,6 +95,7 @@ func (e *NodeExecutor) executeWithRetries( parameters map[string]any, maxRetries, retryDelay, timeout int, startTime time.Time, + view *BranchView, ) error { var lastErr error var lastFunctionCallingOutputs []map[string]any @@ -104,11 +106,11 @@ func (e *NodeExecutor) executeWithRetries( e.finalizeNode(ctx, task, node, nil, lastFunctionCallingOutputs, ctx.Err(), startTime, taskRunnerDomain.StatusCancelled) return ctx.Err() default: - outputs, functionCallingOutputs, err := e.executeOperation(ctx, task, executor, credentials, processedDataList, parameters, timeout) + outputs, functionCallingOutputs, branches, err := e.executeOperation(ctx, task, executor, credentials, processedDataList, parameters, timeout, view) lastFunctionCallingOutputs = functionCallingOutputs if err == nil { - e.storeNodeOutputs(node, outputs) + e.storeNodeOutputs(task.ID, node, outputs, processedDataList, branches, view) e.finalizeNode(ctx, task, node, outputs, functionCallingOutputs, nil, startTime, taskRunnerDomain.StatusSuccess) return nil } @@ -145,7 +147,8 @@ func (e *NodeExecutor) executeOperation( processedDataList []map[string]any, parameters map[string]any, timeout int, -) ([]map[string]any, []map[string]any, error) { + view *BranchView, +) ([]map[string]any, []map[string]any, map[string][]int, error) { attemptCtx := ctx if timeout > 0 { var cancel context.CancelFunc @@ -184,7 +187,7 @@ func (e *NodeExecutor) executeOperation( } for i, data := range processedDataList { - resolvedParameters := e.dataProcessor.ProcessNodeData(parameters, i, triggerData) + resolvedParameters := e.dataProcessor.ProcessNodeData(parameters, i, triggerData, view) for k, v := range resolvedParameters { if v == nil { continue @@ -198,11 +201,16 @@ func (e *NodeExecutor) executeOperation( } } + if branching, ok := executor.(nodeEngineDomainExecutors.BranchingExecutor); ok { + executorOutputs, branches, err := branching.ExecuteBranches(attemptCtx, credentials, processedDataList) + return executorOutputs, functionCallingOutputs, branches, err + } + executorOutputs, err := executor.ExecuteWithContext(attemptCtx, credentials, processedDataList) - return executorOutputs, functionCallingOutputs, err + return executorOutputs, functionCallingOutputs, nil, err } -func (e *NodeExecutor) prepareAndProcessNodeData(task *taskRunnerDomainTask.Task, node *dag.Node) []map[string]any { +func (e *NodeExecutor) prepareAndProcessNodeData(task *taskRunnerDomainTask.Task, node *dag.Node, view *BranchView) []map[string]any { nodeDataMap := make(map[string]any) if node.Instruction != "" { @@ -223,17 +231,17 @@ func (e *NodeExecutor) prepareAndProcessNodeData(task *taskRunnerDomainTask.Task } parentNodes := task.DAG.NodeParents(node.ID) - maxItems := e.calculateMaxItems(task, parentNodes) + maxItems := e.calculateMaxItems(task, parentNodes, view) executionDataList := e.createExecutionDataList(nodeDataMap, maxItems) for i, data := range executionDataList { - executionDataList[i] = e.dataProcessor.ProcessNodeData(data, i, triggerData) + executionDataList[i] = e.dataProcessor.ProcessNodeData(data, i, triggerData, view) } return executionDataList } -func (e *NodeExecutor) calculateMaxItems(task *taskRunnerDomainTask.Task, parentNodes []string) int { +func (e *NodeExecutor) calculateMaxItems(task *taskRunnerDomainTask.Task, parentNodes []string, view *BranchView) int { maxItems := 1 var builder strings.Builder for _, parentID := range parentNodes { @@ -242,7 +250,7 @@ func (e *NodeExecutor) calculateMaxItems(task *taskRunnerDomainTask.Task, parent continue } nodeKey := BuildNodeKeyWithBuilder(&builder, parentNode.NodeID, parentNode.ID) - if outputs, ok := e.outputStore.Get(nodeKey); ok && len(outputs) > maxItems { + if outputs, ok := e.outputStore.Get(view.Task(), view.StoreKey(nodeKey)); ok && len(outputs) > maxItems { maxItems = len(outputs) } } @@ -270,6 +278,7 @@ func (e *NodeExecutor) finalizeNode( status taskRunnerDomain.Status, ) { executionTime := time.Since(startTime).Milliseconds() + task.NodeStatuses.Store(node.ID, status.String()) nodeExecutionID, exists := task.NodeExecutionIDMap[node.ID] if !exists { return @@ -292,9 +301,29 @@ func (e *NodeExecutor) finalizeNode( e.publishNodeEvent(ctx, task, node, status, outputs, errorStr, executionTime) } -func (e *NodeExecutor) storeNodeOutputs(node *dag.Node, outputs []map[string]any) { +func (e *NodeExecutor) storeNodeOutputs( + taskID uuid.UUID, + node *dag.Node, + outputs []map[string]any, + items []map[string]any, + branches map[string][]int, + view *BranchView, +) { nodeKey := BuildNodeKey(node.NodeID, node.ID) - e.outputStore.Store(nodeKey, outputs) + e.outputStore.Store(taskID, nodeKey, outputs) + + for handle, indexes := range branches { + branchItems := make([]map[string]any, 0, len(indexes)) + absolute := make([]int, 0, len(indexes)) + for _, index := range indexes { + if index < 0 || index >= len(items) { + continue + } + branchItems = append(branchItems, items[index]) + absolute = append(absolute, view.AbsoluteIndex(index)) + } + e.outputStore.StoreBranch(taskID, nodeKey, handle, branchItems, absolute) + } } func (e *NodeExecutor) extractSettings(node *dag.Node) (maxRetries, retryDelay, timeout int) { diff --git a/apps/platform-api/internal/taskrunner/application/taskrunner/output_store.go b/apps/platform-api/internal/taskrunner/application/taskrunner/output_store.go index 9a1bfdd..e7965c1 100644 --- a/apps/platform-api/internal/taskrunner/application/taskrunner/output_store.go +++ b/apps/platform-api/internal/taskrunner/application/taskrunner/output_store.go @@ -1,11 +1,25 @@ package taskrunner import ( + "strings" + "sync" "time" + + "github.com/google/uuid" ) type OutputStore struct { - cache *LRUCache + cache *LRUCache + mu sync.RWMutex + indexes map[uuid.UUID]map[string][]int +} + +func scopedKey(taskID uuid.UUID, key string) string { + var builder strings.Builder + builder.WriteString(taskID.String()) + builder.WriteString("|") + builder.WriteString(key) + return builder.String() } func NewOutputStore() *OutputStore { @@ -14,14 +28,40 @@ func NewOutputStore() *OutputStore { defaultTTL = 1 * time.Hour ) return &OutputStore{ - cache: NewLRUCache(defaultMaxSize, defaultTTL), + cache: NewLRUCache(defaultMaxSize, defaultTTL), + indexes: make(map[uuid.UUID]map[string][]int), + } +} + +func (r *OutputStore) Store(taskID uuid.UUID, nodeKey string, outputs []map[string]any) { + r.cache.Store(scopedKey(taskID, nodeKey), outputs) +} + +func (r *OutputStore) Get(taskID uuid.UUID, nodeKey string) ([]map[string]any, bool) { + return r.cache.Get(scopedKey(taskID, nodeKey)) +} + +func (r *OutputStore) StoreBranch(taskID uuid.UUID, nodeKey, handle string, outputs []map[string]any, absolute []int) { + branchKey := BuildBranchKey(nodeKey, handle) + r.cache.Store(scopedKey(taskID, branchKey), outputs) + + r.mu.Lock() + defer r.mu.Unlock() + if r.indexes[taskID] == nil { + r.indexes[taskID] = make(map[string][]int) } + r.indexes[taskID][branchKey] = absolute } -func (r *OutputStore) Store(nodeKey string, outputs []map[string]any) { - r.cache.Store(nodeKey, outputs) +func (r *OutputStore) Indexes(taskID uuid.UUID, branchKey string) ([]int, bool) { + r.mu.RLock() + defer r.mu.RUnlock() + absolute, ok := r.indexes[taskID][branchKey] + return absolute, ok } -func (r *OutputStore) Get(nodeKey string) ([]map[string]any, bool) { - return r.cache.Get(nodeKey) +func (r *OutputStore) Release(taskID uuid.UUID) { + r.mu.Lock() + defer r.mu.Unlock() + delete(r.indexes, taskID) } diff --git a/apps/platform-api/internal/taskrunner/application/taskrunner/task_execution_coordinator.go b/apps/platform-api/internal/taskrunner/application/taskrunner/task_execution_coordinator.go index 857bc88..4cb0651 100644 --- a/apps/platform-api/internal/taskrunner/application/taskrunner/task_execution_coordinator.go +++ b/apps/platform-api/internal/taskrunner/application/taskrunner/task_execution_coordinator.go @@ -12,7 +12,6 @@ import ( credentialOAuthApplicationRegenerate "github.com/blocknextai/platform-api/internal/credentialoauth/application/regenerate" executionsApplicationNodeExecutions "github.com/blocknextai/platform-api/internal/executions/application/nodeexecutions" executionsApplicationTaskexecutions "github.com/blocknextai/platform-api/internal/executions/application/taskexecutions" - "github.com/blocknextai/platform-api/internal/executions/domain/nodeexecutions" nodeEngineDomainExecutors "github.com/blocknextai/platform-api/internal/nodeengine/domain/executors" taskRunnerDomain "github.com/blocknextai/platform-api/internal/taskrunner/domain" taskRunnerDomainTask "github.com/blocknextai/platform-api/internal/taskrunner/domain/task" @@ -45,6 +44,7 @@ func NewTaskExecutionCoordinator( } func (c *taskExecutionCoordinator) ExecuteTask(ctx context.Context, task *taskRunnerDomainTask.Task) error { + defer c.nodeExecutor.outputStore.Release(task.ID) return c.executeTaskNodes(ctx, task) } @@ -52,7 +52,8 @@ func (c *taskExecutionCoordinator) CancelExecution(ctx context.Context, taskID u return nil } -func (c *taskExecutionCoordinator) restoreOutputStore(ctx context.Context, taskID uuid.UUID) { +func (c *taskExecutionCoordinator) seedFromDatabase(ctx context.Context, task *taskRunnerDomainTask.Task) { + taskID := task.ID nodeExecutions, err := c.nodeExecutionService.GetAllByTaskID(ctx, taskID) if err != nil { slog.WarnContext(ctx, "failed to load node executions for output restoration", @@ -64,6 +65,8 @@ func (c *taskExecutionCoordinator) restoreOutputStore(ctx context.Context, taskI restored := 0 for _, nx := range nodeExecutions { + task.NodeStatuses.Store(nx.NodeID, nx.Status) + if nx.Status != taskRunnerDomain.StatusSuccess.String() { continue } @@ -71,7 +74,7 @@ func (c *taskExecutionCoordinator) restoreOutputStore(ctx context.Context, taskI continue } nodeKey := BuildNodeKey(nx.NodeType, nx.NodeID) - c.nodeExecutor.outputStore.Store(nodeKey, nx.Outputs) + c.nodeExecutor.outputStore.Store(taskID, nodeKey, nx.Outputs) restored++ } @@ -88,7 +91,7 @@ func (c *taskExecutionCoordinator) executeTaskNodes(ctx context.Context, task *t return ErrTaskDAGIsEmpty } - c.restoreOutputStore(ctx, task.ID) + c.seedFromDatabase(ctx, task) startNodeIDs := task.DAG.StartNodes() if len(startNodeIDs) == 0 { @@ -146,6 +149,10 @@ func (c *taskExecutionCoordinator) executeNode(ctx context.Context, task *taskRu default: } + if _, started := task.StartedNodes.LoadOrStore(node.ID, struct{}{}); started { + return nil + } + nodeExecutionID, exists := task.NodeExecutionIDMap[node.ID] if !exists { if _, runnable := nodeEngineDomainExecutors.GetExecutor(node.NodeID); !runnable { @@ -154,11 +161,12 @@ func (c *taskExecutionCoordinator) executeNode(ctx context.Context, task *taskRu return ErrNodeExecutionNotFound } - if task.PreviousNodeOutputs != nil { + if task.PreviousNodeOutputs != nil && !routesItems(node) { nodeKey := BuildNodeKey(node.NodeID, node.ID) outputs, hasPreviousOutputs := task.PreviousNodeOutputs[nodeKey] if hasPreviousOutputs { - c.nodeExecutor.outputStore.Store(nodeKey, outputs) + c.nodeExecutor.outputStore.Store(task.ID, nodeKey, outputs) + task.NodeStatuses.Store(node.ID, taskRunnerDomain.StatusSuccess.String()) c.eventPublisher.PublishNodeEvent( ctx, @@ -185,7 +193,7 @@ func (c *taskExecutionCoordinator) executeNode(ctx context.Context, task *taskRu } if node.Settings != nil && node.Settings.Disabled { - c.markNodeAsSkipped(ctx, task, node) + c.markNodeAsSkipped(ctx, task, node, ErrNodeDisabled.Error()) return c.executeChildNodes(ctx, task, node) } @@ -214,22 +222,21 @@ func (c *taskExecutionCoordinator) executeChildNodes(ctx context.Context, task * return nil } - nodeExecutions, err := c.nodeExecutionService.GetAllByTaskID(ctx, task.ID) - if err != nil { - return err - } - - var builder strings.Builder - nodeExecutionsByKey := make(map[string]*nodeexecutions.NodeExecution, len(nodeExecutions)) - for _, ne := range nodeExecutions { - key := BuildNodeKeyWithBuilder(&builder, ne.NodeType, ne.NodeID) - nodeExecutionsByKey[key] = ne - } - readyChildren := make([]*dag.Node, 0) for _, childNodeID := range childNodes { childNode := task.DAG.NodeByID(childNodeID) - if childNode == nil || !c.areParentNodesCompleted(task, childNode, nodeExecutionsByKey) { + if childNode == nil { + continue + } + reached, err := branchReaches(c.nodeExecutor.outputStore, task.ID, task.DAG, node, childNodeID) + if err != nil { + return err + } + if !reached { + c.skipUnreachable(ctx, task, childNode) + continue + } + if !c.areParentNodesCompleted(task, childNode) { continue } readyChildren = append(readyChildren, childNode) @@ -270,7 +277,7 @@ func (c *taskExecutionCoordinator) executeChildNodes(ctx context.Context, task * return nil } -func (c *taskExecutionCoordinator) markNodeAsSkipped(ctx context.Context, task *taskRunnerDomainTask.Task, node *dag.Node) { +func (c *taskExecutionCoordinator) markNodeAsSkipped(ctx context.Context, task *taskRunnerDomainTask.Task, node *dag.Node, reason string) { nodeExecutionID, exists := task.NodeExecutionIDMap[node.ID] if !exists { return @@ -312,9 +319,14 @@ func (c *taskExecutionCoordinator) markNodeAsSkipped(ctx context.Context, task * 0, ) - skipMessage := ErrNodeDisabled.Error() endTime := time.Now().UTC() executionTime := endTime.Sub(startTime).Milliseconds() + task.NodeStatuses.Store(node.ID, taskRunnerDomain.StatusSkipped.String()) + + var errorMessage *string + if strings.TrimSpace(reason) != "" { + errorMessage = new(reason) + } if err := c.nodeExecutionService.Update( ctx, @@ -324,7 +336,7 @@ func (c *taskExecutionCoordinator) markNodeAsSkipped(ctx context.Context, task * nil, nil, nil, - &skipMessage, + errorMessage, nil, &endTime, ); err != nil { @@ -346,7 +358,7 @@ func (c *taskExecutionCoordinator) markNodeAsSkipped(ctx context.Context, task * node.NodeID, taskRunnerDomain.StatusSkipped, nil, - skipMessage, + reason, executionTime, ) } @@ -354,28 +366,23 @@ func (c *taskExecutionCoordinator) markNodeAsSkipped(ctx context.Context, task * func (c *taskExecutionCoordinator) areParentNodesCompleted( task *taskRunnerDomainTask.Task, node *dag.Node, - nodeExecutionsByKey map[string]*nodeexecutions.NodeExecution, ) bool { parentNodes := task.DAG.NodeParents(node.ID) if len(parentNodes) == 0 { return true } - var builder strings.Builder for _, parentNodeID := range parentNodes { parentNode := task.DAG.NodeByID(parentNodeID) if parentNode == nil { continue } - key := BuildNodeKeyWithBuilder(&builder, parentNode.NodeID, parentNode.ID) - - nodeExecution, exists := nodeExecutionsByKey[key] + status, exists := nodeStatus(task, parentNode.ID) if !exists { return false } - status := nodeExecution.Status isCompleted := status == taskRunnerDomain.StatusSuccess.String() || status == taskRunnerDomain.StatusSkipped.String() || (status == taskRunnerDomain.StatusFailed.String() && parentNode.Settings != nil && parentNode.Settings.ContinueOnError) @@ -387,3 +394,60 @@ func (c *taskExecutionCoordinator) areParentNodesCompleted( return true } + +func (c *taskExecutionCoordinator) skipUnreachable( + ctx context.Context, + task *taskRunnerDomainTask.Task, + root *dag.Node, +) { + skipped := make(map[string]bool) + + var walk func(node *dag.Node) + walk = func(node *dag.Node) { + if skipped[node.ID] { + return + } + skipped[node.ID] = true + c.markNodeAsSkipped(ctx, task, node, "") + + for _, childID := range task.DAG.NodeChildren(node.ID) { + child := task.DAG.NodeByID(childID) + if child == nil { + continue + } + if c.hasLiveParent(task, childID, skipped) { + continue + } + walk(child) + } + } + + walk(root) +} + +func (c *taskExecutionCoordinator) hasLiveParent( + task *taskRunnerDomainTask.Task, + nodeID string, + skipped map[string]bool, +) bool { + for _, parentID := range task.DAG.NodeParents(nodeID) { + if skipped[parentID] { + continue + } + if status, ok := nodeStatus(task, parentID); ok && + status == taskRunnerDomain.StatusSkipped.String() { + continue + } + return true + } + return false +} + +func nodeStatus(task *taskRunnerDomainTask.Task, nodeID string) (string, bool) { + value, ok := task.NodeStatuses.Load(nodeID) + if !ok { + return "", false + } + status, ok := value.(string) + return status, ok +} diff --git a/apps/platform-api/internal/taskrunner/application/taskrunner/task_executor.go b/apps/platform-api/internal/taskrunner/application/taskrunner/task_executor.go index f15c1f1..7b0cbc0 100644 --- a/apps/platform-api/internal/taskrunner/application/taskrunner/task_executor.go +++ b/apps/platform-api/internal/taskrunner/application/taskrunner/task_executor.go @@ -27,6 +27,7 @@ type TaskExecutor struct { concurrencyLimitResolver taskRunnerDomainTaskRunner.ConcurrencyLimitResolver workerID string heartbeatInterval time.Duration + maxExecutionTime time.Duration } func NewTaskExecutor( @@ -40,6 +41,7 @@ func NewTaskExecutor( concurrencyLimitResolver taskRunnerDomainTaskRunner.ConcurrencyLimitResolver, workerID string, heartbeatInterval time.Duration, + maxExecutionTime time.Duration, ) *TaskExecutor { return &TaskExecutor{ taskExecutionService: taskExecutionService, @@ -52,6 +54,7 @@ func NewTaskExecutor( concurrencyLimitResolver: concurrencyLimitResolver, workerID: workerID, heartbeatInterval: heartbeatInterval, + maxExecutionTime: maxExecutionTime, } } @@ -304,6 +307,12 @@ func (e *TaskExecutor) runTask(ctx context.Context, task *taskRunnerDomainTask.T taskCtx := e.contextManager.CreateContext(task.ID, ctx) + if e.maxExecutionTime > 0 { + var cancelExecution context.CancelFunc + taskCtx, cancelExecution = context.WithTimeout(taskCtx, e.maxExecutionTime) + defer cancelExecution() + } + if err := e.lifecycleManager.StartTask(taskCtx, task); err != nil { if failErr := e.lifecycleManager.HandleTaskFailure(ctx, task, err); failErr != nil { slog.ErrorContext(ctx, "failed to handle task failure after start error", @@ -315,6 +324,9 @@ func (e *TaskExecutor) runTask(ctx context.Context, task *taskRunnerDomainTask.T } if err := e.executionCoordinator.ExecuteTask(taskCtx, task); err != nil { + if errors.Is(err, context.DeadlineExceeded) { + err = ErrTaskExecutionTimeLimit + } if errors.Is(err, context.Canceled) { if cancelErr := e.lifecycleManager.HandleTaskCancellation(ctx, task); cancelErr != nil { slog.ErrorContext(ctx, "failed to handle task cancellation", diff --git a/apps/platform-api/internal/taskrunner/application/taskrunner/task_lifecycle_manager.go b/apps/platform-api/internal/taskrunner/application/taskrunner/task_lifecycle_manager.go index ed9e91b..082b3cd 100644 --- a/apps/platform-api/internal/taskrunner/application/taskrunner/task_lifecycle_manager.go +++ b/apps/platform-api/internal/taskrunner/application/taskrunner/task_lifecycle_manager.go @@ -210,6 +210,7 @@ func (m *taskLifecycleManager) CreateNodeExecutions(ctx context.Context, task *t return err } task.NodeExecutionIDMap[node.ID] = nodeExecutionID + task.NodeStatuses.Store(node.ID, taskRunnerDomain.StatusSuccess.String()) if err := m.nodeExecutionService.Update( ctx, diff --git a/apps/platform-api/internal/taskrunner/application/taskrunner/value_path.go b/apps/platform-api/internal/taskrunner/application/taskrunner/value_path.go new file mode 100644 index 0000000..e7b693d --- /dev/null +++ b/apps/platform-api/internal/taskrunner/application/taskrunner/value_path.go @@ -0,0 +1,119 @@ +package taskrunner + +import ( + "strconv" + "strings" + + "github.com/blocknextai/go-packages/cast" +) + +func getNestedValueWithArrayAccess(data any, path string) any { + parts := splitPathPreservingParentheses(path) + if len(parts) == 0 { + return data + } + + current := data + part := parts[0] + + if strings.Contains(part, "(") && strings.Contains(part, ")") { + current = getArrayAccessValue(current, part) + } else { + currentMap, ok := current.(map[string]any) + if !ok { + return nil + } + val, ok := currentMap[part] + if !ok { + return nil + } + current = val + } + + if len(parts) == 1 { + return getFirstElementIfSlice(current) + } + + return getNestedValueWithArrayAccess(current, strings.Join(parts[1:], ".")) +} + +func getFirstElementIfSlice(current any) any { + if arr, ok := current.([]any); ok && len(arr) > 0 { + return arr[0] + } + if arr, ok := current.([]map[string]any); ok && len(arr) > 0 { + return arr[0] + } + return current +} + +func splitPathPreservingParentheses(path string) []string { + var parts []string + var current strings.Builder + parenCount := 0 + + for _, char := range path { + switch char { + case '(': + parenCount++ + case ')': + parenCount-- + } + + if char == '.' && parenCount == 0 { + if current.Len() > 0 { + parts = append(parts, current.String()) + current.Reset() + } + } else { + current.WriteRune(char) + } + } + + if current.Len() > 0 { + parts = append(parts, current.String()) + } + + return parts +} + +func getArrayAccessValue(data any, accessor string) any { + methodStart := strings.Index(accessor, "(") + methodEnd := strings.Index(accessor, ")") + + if methodStart == -1 || methodEnd == -1 { + return nil + } + + methodName := accessor[:methodStart] + argsStr := accessor[methodStart+1 : methodEnd] + + slice := cast.ToSlice(data) + if len(slice) == 0 { + return nil + } + + switch methodName { + case "get": + index, err := strconv.Atoi(argsStr) + if err != nil || index < 0 || index >= len(slice) { + return nil + } + return slice[index] + case "first": + return slice[0] + case "last": + return slice[len(slice)-1] + default: + return nil + } +} + +func isComplexType(value any) bool { + switch value.(type) { + case map[string]any, []any, []map[string]any: + return true + default: + return false + } +} diff --git a/apps/platform-api/internal/taskrunner/domain/task/task.go b/apps/platform-api/internal/taskrunner/domain/task/task.go index 8dab79b..2a0cc7d 100644 --- a/apps/platform-api/internal/taskrunner/domain/task/task.go +++ b/apps/platform-api/internal/taskrunner/domain/task/task.go @@ -1,6 +1,7 @@ package task import ( + "sync" "time" "github.com/blocknextai/go-packages/dag" @@ -23,6 +24,8 @@ type Task struct { StartTime *time.Time EndTime *time.Time NodeExecutionIDMap map[string]uuid.UUID + StartedNodes sync.Map + NodeStatuses sync.Map PreviousNodeOutputs map[string][]map[string]any TriggerContext *nodeEngineDomainAdapters.TriggerContext diff --git a/apps/platform-api/internal/taskrunner/module.go b/apps/platform-api/internal/taskrunner/module.go index 060aa9a..b41b12b 100644 --- a/apps/platform-api/internal/taskrunner/module.go +++ b/apps/platform-api/internal/taskrunner/module.go @@ -123,6 +123,7 @@ func NewModule(deps Dependencies) (*Module, error) { concurrencyLimitResolver, workerID, deps.TaskRunnerOptions.HeartbeatInterval, + deps.TaskRunnerOptions.MaxExecutionTime, ) dispatcher, err := taskRunnerInfrastructure.NewDispatcher( diff --git a/apps/platform-api/internal/workflows/README.md b/apps/platform-api/internal/workflows/README.md index e3f5e7b..c060556 100644 --- a/apps/platform-api/internal/workflows/README.md +++ b/apps/platform-api/internal/workflows/README.md @@ -25,11 +25,14 @@ This context is the system of record for workflow definitions scoped to an organ | `$_.` | Field from the referenced node's output for the current item (node key = catalog `nodeId` + `_` + canvas `id`, e.g. `$gemini.imagen_2.images`) | | `$[0].` / `$[*].` | Explicit index / all items collected into an array | | `$.first().` / `.last()` / `.get(n)` | Positional access over the node's output list | +| `$input.` | The item feeding this node at the same position, without naming the node it comes from; only defined when the node has exactly one incoming edge | | `$trigger.source` / `.sender` / `.prompt` / `.payload` | Fields of the run's `TriggerContext` (webhook adapter output or runtime prompt) | **Credentials are referenced, never stored.** The workflow JSON carries no secrets: a node's `credentials` map holds opaque reference strings of the form `credential::` (`internal/common/domain/credential`). At execution time taskrunner's `CredentialProcessor` parses the reference, resolves the owner scope, and fetches (refreshing OAuth tokens if needed) the actual credential material — which therefore lives only in the credentials store, and a duplicated or exported workflow leaks nothing. -**Deliberate simplicity.** Control flow is intentionally minimal: the only system nodes are `starter`, `condition` (boolean branch — edges carry `condition: "true"|"false"` and `dag.ConditionalChildren` picks the branch from the node's `status` output), and `sleep`. There is **no loop, no sub-workflow, and no generic HTTP node type**, and the graph must be acyclic — `dag.New` runs a topological sort and rejects cycles (`ErrCycleDetected`). +**Deliberate simplicity.** Control flow is intentionally minimal: the only system nodes are `starter`, `condition`, `sleep` and the canvas-only `annotation`. There is **no loop, no sub-workflow, and no generic HTTP node type**, and the graph must be acyclic — `dag.New` runs a topological sort and rejects cycles (`ErrCycleDetected`). + +**Branching is per item.** A node that routes implements `executors.BranchingExecutor`, returning the indexes of the items that leave through each output handle; `condition` returns them under `true` and `false`. The runner stores each branch separately (`#`) together with the original index of every item it kept, so a consumer reads the branch its own edge leaves from, and a `$reference` to a node *upstream* of the branch still resolves to the matching item rather than to position `i` of the filtered list. A handle with no items reaches nothing: its children are marked `skipped`, and the skip walks on to any descendant whose every parent is skipped — a join with one live parent still runs. An edge with no `sourceHandle` is not routed, so flows authored before handles existed keep their old behaviour. ## AI workflow generation diff --git a/apps/platform-api/prompts/workflow-generation-system-instruction.md b/apps/platform-api/prompts/workflow-generation-system-instruction.md index 771dfc6..25dc05c 100644 --- a/apps/platform-api/prompts/workflow-generation-system-instruction.md +++ b/apps/platform-api/prompts/workflow-generation-system-instruction.md @@ -37,6 +37,8 @@ RIGHT (nodes + edges arrays): } NEVER omit the "edges" array. Even a single-executable-node flow has at least one edge: starter → first executable node. + +An edge leaving a node that declares more than one output MUST name which one it leaves from with "sourceHandle". `system_condition` declares "true" and "false": {"id":"xy-edge__1-2","source":"1","sourceHandle":"true","target":"2"}. Omit it and both branches run. Branching is per item — a condition fed ten items sends each one down the branch its own comparison chose — so a node after a condition should reference the condition or a node before it, and the runner lines the items up. NEVER emit nodes as an object map keyed by id. Always an ordered array. === NODE STRUCTURE (FLAT, NO data FIELD) === @@ -173,6 +175,8 @@ A reference can appear in two places: Rule: If a consumer node has incoming edges, the upstream output MUST be referenced — either via parameters (preferred) or instruction. The edge alone is not enough; the runtime resolves the dependency from the reference. +$input. is shorthand for "the item feeding this node at the same position", so a node with exactly one incoming edge can read its input without naming the node it comes from: "parameters": { "text": "$input.summary" }. It is undefined for a node with several incoming edges — name the node there. Prefer the shorthand on a straight chain; it survives the upstream node being renamed or replaced. + === TRIGGER VARIABLES (WEBHOOK FLOWS) === When the flow is intended to run via webhook (telegram, slack, discord, whatsapp, generic), the trigger payload is injected into the task. You can reference the payload inside any node's instruction using these variables (always quoted): diff --git a/apps/platform/public/locales/en/ui.json b/apps/platform/public/locales/en/ui.json index d429a22..1f55c5a 100644 --- a/apps/platform/public/locales/en/ui.json +++ b/apps/platform/public/locales/en/ui.json @@ -815,6 +815,7 @@ "ui.text.submit": "Submit", "ui.text.success": "Success", "ui.text.successful": "Successful", + "ui.text.skipped": "Skipped", "ui.text.summary": "Summary", "ui.text.summaryPlaceholder": "Brief description that appears in search results", "ui.text.symbol": "Symbol", diff --git a/apps/platform/src/features/flow-editor/components/flow-canvas.tsx b/apps/platform/src/features/flow-editor/components/flow-canvas.tsx index 6b067c8..b9d2cf4 100644 --- a/apps/platform/src/features/flow-editor/components/flow-canvas.tsx +++ b/apps/platform/src/features/flow-editor/components/flow-canvas.tsx @@ -203,7 +203,8 @@ const FlowCanvas = ({ const apiNode = apiNodes.find((a) => a.id === node.nodeId) return { id: node.id, - title: apiNode?.name || node.nodeId, + title: node.title || apiNode?.name || node.nodeId, + catalogTitle: apiNode?.name || node.nodeId, description: apiNode?.description || '', tags: apiNode?.tags || [], category: apiNode?.category || '', diff --git a/apps/platform/src/features/flow-editor/hooks/use-flow-save.ts b/apps/platform/src/features/flow-editor/hooks/use-flow-save.ts index 0338571..7f90779 100644 --- a/apps/platform/src/features/flow-editor/hooks/use-flow-save.ts +++ b/apps/platform/src/features/flow-editor/hooks/use-flow-save.ts @@ -82,6 +82,7 @@ export function useFlowSave({ initialFlow, nodes, edges }: UseFlowSaveOptions) { const stripNodesForSave = (nodeList: FlowNode[]) => nodeList.map(({ data, ...node }) => ({ ...node, + title: data?.title === data?.catalogTitle ? undefined : data?.title, handleLayout: data?.handleLayout, parameters: data?.note === undefined diff --git a/apps/platform/src/features/organizations/components/history/run-step-list.tsx b/apps/platform/src/features/organizations/components/history/run-step-list.tsx index 2242173..2205c2b 100644 --- a/apps/platform/src/features/organizations/components/history/run-step-list.tsx +++ b/apps/platform/src/features/organizations/components/history/run-step-list.tsx @@ -1,7 +1,7 @@ import { useTranslation } from 'react-i18next' import { RunStepDetail } from './run-step-detail' import { Card, CardContent, CardHeader, CardTitle } from '@/components/ui/card' -import { Play, CheckCircle2, XCircle } from 'lucide-react' +import { Play, CheckCircle2, XCircle, SkipForward } from 'lucide-react' const RunStepList = ({ nodeExecutions, @@ -16,6 +16,8 @@ const RunStepList = ({ nodeExecutions?.filter((n) => n.status === 'success').length || 0 const failedNodes = nodeExecutions?.filter((n) => n.status === 'failed').length || 0 + const skippedNodes = + nodeExecutions?.filter((n) => n.status === 'skipped').length || 0 return ( <> @@ -27,7 +29,7 @@ const RunStepList = ({ -
+
@@ -63,6 +65,19 @@ const RunStepList = ({
+
+
+ +
+
+
+ {t('ui.text.skipped')} +
+
+ {skippedNodes} +
+
+
diff --git a/packages/go-packages/dag/node.go b/packages/go-packages/dag/node.go index c66ac93..bf17577 100644 --- a/packages/go-packages/dag/node.go +++ b/packages/go-packages/dag/node.go @@ -2,12 +2,14 @@ package dag // Node represents a single workflow node in the DAG, including its type, // instructions, parameters, settings, credentials, and canvas placement. -// HandleLayout is canvas-only: it names the sides the node's input and output -// handles sit on, in "-" form ("l-r", "t-b", …). +// Title is the name the author gave the node on the canvas; empty means the +// catalog name. HandleLayout is canvas-only: it names the sides the node's +// input and output handles sit on, in "-" form ("l-r", "t-b", …). type Node struct { ID string `json:"id"` Type string `json:"type"` NodeID string `json:"nodeId"` + Title string `json:"title,omitempty"` Instruction string `json:"instruction,omitempty"` RuntimeInstruction string `json:"runtimeInstruction,omitempty"` RuntimePrompt string `json:"runtimePrompt,omitempty"`