agent-framework/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/ForeachExecutor.cs

// Copyright (c) Microsoft. All rights reserved.

using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Agents.AI.Workflows.Declarative.Extensions;
using Microsoft.Agents.AI.Workflows.Declarative.Interpreter;
using Microsoft.Agents.AI.Workflows.Declarative.PowerFx;
using Microsoft.Agents.ObjectModel;
using Microsoft.Agents.ObjectModel.Abstractions;
using Microsoft.PowerFx.Types;
using Microsoft.Shared.Diagnostics;

namespace Microsoft.Agents.AI.Workflows.Declarative.ObjectModel;

internal sealed class ForeachExecutor : DeclarativeActionExecutor<Foreach>
{
    public static class Steps
    {
        public static string Start(string id) => $"{id}_{nameof(Start)}";
        public static string Next(string id) => $"{id}_{nameof(Next)}";
        public static string End(string id) => $"{id}_{nameof(End)}";
    }

    private int _index;
    private FormulaValue[] _values;

    public ForeachExecutor(Foreach model, WorkflowFormulaState state)
        : base(model, state)
    {
        this._values = [];
    }

    public bool HasValue { get; private set; }

    protected override bool IsDiscreteAction => false;

    protected override async ValueTask<object?> ExecuteAsync(IWorkflowContext context, CancellationToken cancellationToken = default)
    {
        Throw.IfNull(this.Model.Items, $"{nameof(this.Model)}.{nameof(this.Model.Items)}");

        this._index = 0;

        EvaluationResult<DataValue> expressionResult = this.Evaluator.GetValue(this.Model.Items);
        if (expressionResult.Value is TableDataValue tableValue)
        {
            this._values = [.. tableValue.Values.Select(value => value.Properties.Values.First().ToFormula())];
        }
        else
        {
            this._values = [expressionResult.Value.ToFormula()];
        }

        await this.ResetStateAsync(context, cancellationToken).ConfigureAwait(false);

        return default;
    }

    public async ValueTask TakeNextAsync(IWorkflowContext context, object? _, CancellationToken cancellationToken)
    {
        if (this.HasValue = this._index < this._values.Length)
        {
            FormulaValue value = this._values[this._index];

            await context.QueueStateUpdateAsync(Throw.IfNull(this.Model.Value), value, cancellationToken).ConfigureAwait(false);

            if (this.Model.Index is not null)
            {
                await context.QueueStateUpdateAsync(this.Model.Index.Path, FormulaValue.New(this._index), cancellationToken).ConfigureAwait(false);
            }

            this._index++;
        }
    }

    public async ValueTask CompleteAsync(IWorkflowContext context, object? _, CancellationToken cancellationToken)
    {
        try
        {
            await this.ResetStateAsync(context, cancellationToken).ConfigureAwait(false);
        }
        finally
        {
            await context.RaiseCompletionEventAsync(this.Model, cancellationToken).ConfigureAwait(false);
        }
    }

    private async Task ResetStateAsync(IWorkflowContext context, CancellationToken cancellationToken)
    {
        await context.QueueStateResetAsync(Throw.IfNull(this.Model.Value), cancellationToken).ConfigureAwait(false);
        if (this.Model.Index is not null)
        {
            await context.QueueStateResetAsync(this.Model.Index, cancellationToken).ConfigureAwait(false);
        }
    }
}