| 738 | func (s *StepsGroup) Config() *StepConfig { return s.config } |
| 739 | |
| 740 | func (s *StepsGroup) Execute(ctx context.Context, input *StepInput) *stream.Reader[*StepOutput] { |
| 741 | reader, writer := stream.Pipe[*StepOutput](1) |
| 742 | |
| 743 | go func() { |
| 744 | defer writer.Close() |
| 745 | startTime := time.Now() |
| 746 | |
| 747 | var outputs []*StepOutput |
| 748 | var lastOutput *StepOutput |
| 749 | |
| 750 | for _, step := range s.steps { |
| 751 | stepInput := &StepInput{ |
| 752 | Input: input.Input, |
| 753 | PreviousStepOutputs: input.PreviousStepOutputs, |
| 754 | AdditionalData: input.AdditionalData, |
| 755 | SessionState: input.SessionState, |
| 756 | } |
| 757 | |
| 758 | if lastOutput != nil { |
| 759 | stepInput.PreviousStepContent = lastOutput.Content |
| 760 | } |
| 761 | |
| 762 | var stepOutput *StepOutput |
| 763 | stepReader := step.Execute(ctx, stepInput) |
| 764 | for { |
| 765 | output, err := stepReader.Recv() |
| 766 | if err != nil { |
| 767 | if errors.Is(err, io.EOF) { |
| 768 | break |
| 769 | } |
| 770 | errorOutput := &StepOutput{ |
| 771 | StepID: s.id, |
| 772 | StepName: s.name, |
| 773 | StepType: StepTypeSteps, |
| 774 | Error: err, |
| 775 | StartTime: startTime, |
| 776 | EndTime: time.Now(), |
| 777 | NestedSteps: outputs, |
| 778 | Metadata: map[string]any{"completed": len(outputs), "total": len(s.steps)}, |
| 779 | } |
| 780 | errorOutput.Duration = errorOutput.EndTime.Sub(errorOutput.StartTime).Seconds() |
| 781 | writer.Send(errorOutput, err) |
| 782 | return |
| 783 | } |
| 784 | stepOutput = output |
| 785 | } |
| 786 | |
| 787 | outputs = append(outputs, stepOutput) |
| 788 | lastOutput = stepOutput |
| 789 | |
| 790 | if ctx.Err() != nil { |
| 791 | writer.Send(nil, ctx.Err()) |
| 792 | return |
| 793 | } |
| 794 | } |
| 795 | |
| 796 | output := &StepOutput{ |
| 797 | StepID: s.id, |