CyberStrikeAI/internal/workflow/join.go

package workflow

import (
	"fmt"
	"strings"
)

const (
	JoinAllMerge      = "all_merge"
	JoinLastByCanvas  = "last_by_canvas"
	JoinFirstNonEmpty = "first_non_empty"
	JoinFailFast      = "fail_fast"
)

var allowedJoinStrategies = map[string]bool{
	JoinAllMerge:      true,
	JoinLastByCanvas:  true,
	JoinFirstNonEmpty: true,
	JoinFailFast:      true,
}

func joinStrategy(node graphNode) string {
	strategy := strings.ToLower(cfgString(node.Config, "join_strategy"))
	if strategy == "" {
		return JoinAllMerge
	}
	return strategy
}

func prepareNodeInputState(rt *workflowRuntime, node graphNode) error {
	if rt == nil || rt.idx == nil || rt.state == nil {
		return nil
	}
	incoming := rt.idx.incoming[node.ID]
	if len(incoming) <= 1 {
		return nil
	}
	strategy := joinStrategy(node)
	if !allowedJoinStrategies[strategy] {
		return fmt.Errorf("节点「%s」使用了未知汇聚策略: %s", firstNonEmpty(node.Label, node.ID), strategy)
	}
	upstreams := make([]map[string]any, 0, len(incoming))
	for _, edge := range incoming {
		out := rt.state.NodeOutputs[edge.Source]
		if out == nil {
			continue
		}
		if isFailedNodeOutput(out) && strategy == JoinFailFast {
			return fmt.Errorf("上游节点「%s」失败，汇聚策略 fail_fast 中止", edge.Source)
		}
		upstreams = append(upstreams, out)
	}
	if len(upstreams) == 0 {
		return nil
	}
	rt.state.LastOutput = mergeUpstreamOutputs(strategy, upstreams)
	return nil
}

func mergeUpstreamOutputs(strategy string, upstreams []map[string]any) map[string]any {
	switch strategy {
	case JoinLastByCanvas:
		return cloneNodeOutput(upstreams[len(upstreams)-1])
	case JoinFirstNonEmpty:
		for _, out := range upstreams {
			if !isEmptyOutputValue(out["output"]) {
				return cloneNodeOutput(out)
			}
		}
		return cloneNodeOutput(upstreams[0])
	default:
		merged := map[string]any{
			"kind":      "join",
			"strategy":  strategy,
			"upstreams": upstreams,
		}
		values := make([]any, 0, len(upstreams))
		for _, out := range upstreams {
			values = append(values, out["output"])
			for k, v := range out {
				if _, exists := merged[k]; !exists {
					merged[k] = v
				}
			}
		}
		merged["output"] = values
		return merged
	}
}

func cloneNodeOutput(in map[string]any) map[string]any {
	out := make(map[string]any, len(in))
	for k, v := range in {
		out[k] = v
	}
	return out
}

func isEmptyOutputValue(v any) bool {
	if v == nil {
		return true
	}
	return strings.TrimSpace(fmt.Sprint(v)) == ""
}

func isFailedNodeOutput(out map[string]any) bool {
	if out == nil {
		return false
	}
	if v, ok := out["error"]; ok && strings.TrimSpace(fmt.Sprint(v)) != "" {
		return true
	}
	if v, ok := out["is_error"]; ok {
		return strings.EqualFold(fmt.Sprint(v), "true")
	}
	return false
}