Method executeParallel

pkg/executionplan/executor.go:177–279

executeParallel 并行执行无依赖的步骤

(ctx context.Context, plan *ExecutionPlan, toolCtx *tools.ToolContext)

// executeParallel 并行执行无依赖的步骤
func (e *Executor) executeParallel(ctx context.Context, plan *ExecutionPlan, toolCtx *tools.ToolContext) error {
	// 构建步骤依赖图
	completed := make(map[string]bool)
	var mu sync.Mutex
	var wg sync.WaitGroup
	var firstError error
	var errMu sync.Mutex

	maxParallel := plan.Options.MaxParallelSteps
	if maxParallel <= 0 {
		maxParallel = 3 // 默认最多3个并行
	}
	sem := make(chan struct{}, maxParallel)

	for {
		// 找出所有可以执行的步骤
		var readySteps []*Step
		mu.Lock()
		for i := range plan.Steps {
			step := &plan.Steps[i]
			if step.Status != StepStatusPending {
				continue
			}
			if e.checkDependenciesWithMap(step, completed) {
				readySteps = append(readySteps, step)
			}
		}
		mu.Unlock()

		// 如果没有可执行的步骤，检查是否全部完成
		if len(readySteps) == 0 {
			// 检查是否还有未完成的步骤
			allDone := true
			for i := range plan.Steps {
				if plan.Steps[i].Status == StepStatusPending || plan.Steps[i].Status == StepStatusRunning {
					allDone = false
					break
				}
			}
			if allDone {
				break
			}
			// 等待正在执行的步骤完成
			time.Sleep(100 * time.Millisecond)
			continue
		}

		// 并行执行准备好的步骤
		for _, step := range readySteps {
			select {
			case <-ctx.Done():
				return ctx.Err()
			case sem <- struct{}{}:
			}

			wg.Add(1)
			go func(s *Step) {
				defer wg.Done()