GrpcDurableTaskClient.cs

// Copyright (c) Microsoft Corporation.
// Licensed under the MIT License.

using System.Collections.Immutable;
using System.Diagnostics;
using System.Text;
using DurableTask.Core.Exceptions;
using DurableTask.Core.History;
using Google.Protobuf.WellKnownTypes;
using Microsoft.DurableTask.Client.Entities;
using Microsoft.DurableTask.Tracing;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using static Microsoft.DurableTask.Protobuf.TaskHubSidecarService;
using P = Microsoft.DurableTask.Protobuf;

namespace Microsoft.DurableTask.Client.Grpc;

/// <summary>
/// Durable Task client implementation that uses gRPC to connect to a remote "sidecar" process.
/// </summary>
public sealed class GrpcDurableTaskClient : DurableTaskClient
{
    readonly ILogger logger;
    readonly TaskHubSidecarServiceClient sidecarClient;
    readonly GrpcDurableTaskClientOptions options;
    readonly DurableEntityClient? entityClient;
    AsyncDisposable asyncDisposable;

    /// <summary>
    /// Initializes a new instance of the <see cref="GrpcDurableTaskClient"/> class.
    /// </summary>
    /// <param name="name">The name of the client.</param>
    /// <param name="options">The gRPC client options.</param>
    /// <param name="logger">The logger.</param>
    [ActivatorUtilitiesConstructor]
    public GrpcDurableTaskClient(
        string name, IOptionsMonitor<GrpcDurableTaskClientOptions> options, ILogger<GrpcDurableTaskClient> logger)
        : this(name, Check.NotNull(options).Get(name), logger)
    {
    }

    /// <summary>
    /// Initializes a new instance of the <see cref="GrpcDurableTaskClient"/> class.
    /// </summary>
    /// <param name="name">The name of the client.</param>
    /// <param name="options">The gRPC client options.</param>
    /// <param name="logger">The logger.</param>
    public GrpcDurableTaskClient(string name, GrpcDurableTaskClientOptions options, ILogger logger)
        : base(name)
    {
        this.logger = Check.NotNull(logger);
        this.options = Check.NotNull(options);
        this.asyncDisposable = GetCallInvoker(options, out CallInvoker callInvoker);
        this.sidecarClient = new TaskHubSidecarServiceClient(callInvoker);

        if (this.options.EnableEntitySupport)
        {
            this.entityClient = new GrpcDurableEntityClient(this.Name, this.DataConverter, this.sidecarClient, logger);
        }
    }

    /// <inheritdoc/>
    public override DurableEntityClient Entities => this.entityClient
        ?? throw new NotSupportedException($"Durable entities are disabled because {nameof(DurableTaskClientOptions)}.{nameof(DurableTaskClientOptions.EnableEntitySupport)}=false");

    DataConverter DataConverter => this.options.DataConverter;

    /// <inheritdoc/>
    public override ValueTask DisposeAsync()
    {
        return this.asyncDisposable.DisposeAsync();
    }

    /// <inheritdoc/>
    // The behavior of this method when the dedupe statuses field is null depends on the server-side implementation.
    public override async Task<string> ScheduleNewOrchestrationInstanceAsync(
        TaskName orchestratorName,
        object? input = null,
        StartOrchestrationOptions? options = null,
        CancellationToken cancellation = default)
    {
        Check.NotEntity(this.options.EnableEntitySupport, options?.InstanceId);

        // We're explicitly OK with an empty version from the options as that had to be explicitly set. It should take precedence over the default.
        string? version = null;
        if (options?.Version is { } v)
        {
            version = v;
        }
        else if (!string.IsNullOrEmpty(this.options.DefaultVersion))
        {
            version = this.options.DefaultVersion;
        }

        P.CreateInstanceRequest request = new()
        {
            Name = orchestratorName.Name,
            Version = version,
            InstanceId = options?.InstanceId ?? Guid.NewGuid().ToString("N"),
            Input = this.DataConverter.Serialize(input),
            RequestTime = DateTimeOffset.UtcNow.ToTimestamp(),
        };

        // Add tags to the collection
        if (request?.Tags != null && options?.Tags != null)
        {
            foreach (KeyValuePair<string, string> tag in options.Tags)
            {
                request.Tags.Add(tag.Key, tag.Value);
            }
        }

        DateTimeOffset? startAt = options?.StartAt;
        this.logger.SchedulingOrchestration(
            request.InstanceId ?? string.Empty,
            orchestratorName,
            sizeInBytes: request.Input != null ? Encoding.UTF8.GetByteCount(request.Input) : 0,
            startAt.GetValueOrDefault(DateTimeOffset.UtcNow));

        if (startAt.HasValue)
        {
            // Convert timestamps to UTC if not already UTC
            request.ScheduledStartTimestamp = Timestamp.FromDateTimeOffset(startAt.Value.ToUniversalTime());
        }

        // Set orchestration ID reuse policy for deduplication support
        if (options?.DedupeStatuses != null)
        {
            // Parse and validate all status strings to enum first
            ImmutableHashSet<OrchestrationRuntimeStatus> dedupeStatuses = options.DedupeStatuses
                .Select(s =>
                {
                    if (!System.Enum.TryParse<OrchestrationRuntimeStatus>(s, ignoreCase: true, out OrchestrationRuntimeStatus status))
                    {
                        throw new ArgumentException(
                            $"Invalid orchestration runtime status: '{s}' for deduplication.");
                    }

                    return status;
                }).ToImmutableHashSet();

            // Convert dedupe statuses to protobuf statuses and create reuse policy
            IEnumerable<P.OrchestrationStatus> dedupeStatusesProto = dedupeStatuses.Select(s => s.ToGrpcStatus());
            request.OrchestrationIdReusePolicy = ProtoUtils.ConvertDedupeStatusesToReusePolicy(dedupeStatusesProto);
        }

        using Activity? newActivity = TraceHelper.StartActivityForNewOrchestration(request);

        try
        {
            P.CreateInstanceResponse? result = await this.sidecarClient.StartInstanceAsync(
            request, cancellationToken: cancellation);
            return result.InstanceId;
        }
        catch (RpcException e) when (e.StatusCode == StatusCode.AlreadyExists)
        {
            throw new OrchestrationAlreadyExistsException(e.Status.Detail);
        }
        catch (RpcException e) when (e.StatusCode == StatusCode.InvalidArgument)
        {
            throw new ArgumentException(e.Status.Detail);
        }
        catch (RpcException e) when (e.StatusCode == StatusCode.Cancelled)
        {
            throw new OperationCanceledException(
                $"The {nameof(this.ScheduleNewOrchestrationInstanceAsync)} operation was canceled.", e, cancellation);
        }
    }

    /// <inheritdoc/>
    public override async Task RaiseEventAsync(
        string instanceId, string eventName, object? eventPayload = null, CancellationToken cancellation = default)
    {
        Check.NotNullOrEmpty(instanceId);
        Check.NotNullOrEmpty(eventName);

        Check.NotEntity(this.options.EnableEntitySupport, instanceId);

        P.RaiseEventRequest request = new()
        {
            InstanceId = instanceId,
            Name = eventName,
            Input = this.DataConverter.Serialize(eventPayload),
        };

        using Activity? traceActivity = TraceHelper.StartActivityForNewEventRaisedFromClient(request, instanceId);

        await this.sidecarClient.RaiseEventAsync(request, cancellationToken: cancellation);
    }

    /// <inheritdoc/>
    public override async Task TerminateInstanceAsync(
        string instanceId, TerminateInstanceOptions? options = null, CancellationToken cancellation = default)
    {
        object? output = options?.Output;
        bool recursive = options?.Recursive ?? false;

        Check.NotNullOrEmpty(instanceId);
        Check.NotEntity(this.options.EnableEntitySupport, instanceId);

        this.logger.TerminatingInstance(instanceId);

        string? serializedOutput = this.DataConverter.Serialize(output);
        await this.sidecarClient.TerminateInstanceAsync(
            new P.TerminateRequest
            {
                InstanceId = instanceId,
                Output = serializedOutput,
                Recursive = recursive,
            },
            cancellationToken: cancellation);
    }

    /// <inheritdoc/>
    public override async Task SuspendInstanceAsync(
        string instanceId, string? reason = null, CancellationToken cancellation = default)
    {
        Check.NotNullOrEmpty(instanceId);
        Check.NotEntity(this.options.EnableEntitySupport, instanceId);

        P.SuspendRequest request = new()
        {
            InstanceId = instanceId,
            Reason = reason,
        };

        try
        {
            await this.sidecarClient.SuspendInstanceAsync(request, cancellationToken: cancellation);
        }
        catch (RpcException e) when (e.StatusCode == StatusCode.Cancelled)
        {
            throw new OperationCanceledException(
                $"The {nameof(this.SuspendInstanceAsync)} operation was canceled.", e, cancellation);
        }
    }

    /// <inheritdoc/>
    public override async Task ResumeInstanceAsync(
        string instanceId, string? reason = null, CancellationToken cancellation = default)
    {
        Check.NotNullOrEmpty(instanceId);
        Check.NotEntity(this.options.EnableEntitySupport, instanceId);

        P.ResumeRequest request = new()
        {
            InstanceId = instanceId,
            Reason = reason,
        };

        try
        {
            await this.sidecarClient.ResumeInstanceAsync(request, cancellationToken: cancellation);
        }
        catch (RpcException e) when (e.StatusCode == StatusCode.Cancelled)
        {
            throw new OperationCanceledException(
                $"The {nameof(this.ResumeInstanceAsync)} operation was canceled.", e, cancellation);
        }
    }

    /// <inheritdoc/>
    public override async Task<OrchestrationMetadata?> GetInstancesAsync(
        string instanceId, bool getInputsAndOutputs = false, CancellationToken cancellation = default)
    {
        Check.NotEntity(this.options.EnableEntitySupport, instanceId);

        if (string.IsNullOrEmpty(instanceId))
        {
            throw new ArgumentNullException(nameof(instanceId));
        }

        P.GetInstanceResponse response = await this.sidecarClient.GetInstanceAsync(
            new P.GetInstanceRequest
            {
                InstanceId = instanceId,
                GetInputsAndOutputs = getInputsAndOutputs,
            },
            cancellationToken: cancellation);

        // REVIEW: Should we return a non-null value instead of !exists?
        if (!response.Exists)
        {
            return null;
        }

        return this.CreateMetadata(response.OrchestrationState, getInputsAndOutputs);
    }

    /// <inheritdoc/>
    public override AsyncPageable<OrchestrationMetadata> GetAllInstancesAsync(OrchestrationQuery? filter = null)
    {
        Check.NotEntity(this.options.EnableEntitySupport, filter?.InstanceIdPrefix);

        return Pageable.Create(async (continuation, pageSize, cancellation) =>
        {
            P.QueryInstancesRequest request = new()
            {
                Query = new P.InstanceQuery
                {
                    CreatedTimeFrom = filter?.CreatedFrom?.ToTimestamp(),
                    CreatedTimeTo = filter?.CreatedTo?.ToTimestamp(),
                    FetchInputsAndOutputs = filter?.FetchInputsAndOutputs ?? false,
                    InstanceIdPrefix = filter?.InstanceIdPrefix,
                    MaxInstanceCount = pageSize ?? filter?.PageSize ?? OrchestrationQuery.DefaultPageSize,
                    ContinuationToken = continuation ?? filter?.ContinuationToken,
                },
            };

            if (filter?.Statuses is not null)
            {
                request.Query.RuntimeStatus.AddRange(filter.Statuses.Select(x => x.ToGrpcStatus()));
            }

            if (filter?.TaskHubNames is not null)
            {
                request.Query.TaskHubNames.AddRange(filter.TaskHubNames);
            }

            try
            {
                P.QueryInstancesResponse response = await this.sidecarClient.QueryInstancesAsync(
                    request, cancellationToken: cancellation);

                bool getInputsAndOutputs = filter?.FetchInputsAndOutputs ?? false;
                IReadOnlyList<OrchestrationMetadata> values = response.OrchestrationState
                    .Select(x => this.CreateMetadata(x, getInputsAndOutputs))
                    .ToList();

                return new Page<OrchestrationMetadata>(values, response.ContinuationToken);
            }
            catch (RpcException e) when (e.StatusCode == StatusCode.Cancelled)
            {
                throw new OperationCanceledException(
                    $"The {nameof(this.GetInstancesAsync)} operation was canceled.", e, cancellation);
            }
        });
    }

    /// <inheritdoc/>
    public override async Task<OrchestrationMetadata> WaitForInstanceStartAsync(
        string instanceId, bool getInputsAndOutputs = false, CancellationToken cancellation = default)
    {
        Check.NotEntity(this.options.EnableEntitySupport, instanceId);

        this.logger.WaitingForInstanceStart(instanceId, getInputsAndOutputs);

        P.GetInstanceRequest request = new()
        {
            InstanceId = instanceId,
            GetInputsAndOutputs = getInputsAndOutputs,
        };

        try
        {
            P.GetInstanceResponse response = await this.sidecarClient.WaitForInstanceStartAsync(
                request, cancellationToken: cancellation);
            return this.CreateMetadata(response.OrchestrationState, getInputsAndOutputs);
        }
        catch (RpcException e) when (e.StatusCode == StatusCode.Cancelled)
        {
            throw new OperationCanceledException(
                $"The {nameof(this.WaitForInstanceStartAsync)} operation was canceled.", e, cancellation);
        }
    }

    /// <inheritdoc/>
    public override async Task<Page<string>> ListInstanceIdsAsync(
        IEnumerable<OrchestrationRuntimeStatus>? runtimeStatus = null,
        DateTimeOffset? completedTimeFrom = null,
        DateTimeOffset? completedTimeTo = null,
        int pageSize = OrchestrationQuery.DefaultPageSize,
        string? lastInstanceKey = null,
        CancellationToken cancellation = default)
    {
        Check.NotEntity(this.options.EnableEntitySupport, null);

        P.ListInstanceIdsRequest request = new()
        {
            PageSize = pageSize,
            LastInstanceKey = lastInstanceKey ?? string.Empty,
        };

        if (runtimeStatus != null)
        {
            request.RuntimeStatus.AddRange(runtimeStatus.Select(x => x.ToGrpcStatus()));
        }

        if (completedTimeFrom.HasValue)
        {
            request.CompletedTimeFrom = completedTimeFrom.Value.ToTimestamp();
        }

        if (completedTimeTo.HasValue)
        {
            request.CompletedTimeTo = completedTimeTo.Value.ToTimestamp();
        }

        try
        {
            P.ListInstanceIdsResponse response = await this.sidecarClient.ListInstanceIdsAsync(
                request,
                cancellationToken: cancellation);

            return new Page<string>(response.InstanceIds.ToList(), response.LastInstanceKey);
        }
        catch (RpcException e) when (e.StatusCode == StatusCode.Cancelled)
        {
            throw new OperationCanceledException(
                $"The {nameof(this.ListInstanceIdsAsync)} operation was canceled.", e, cancellation);
        }
    }

    /// <inheritdoc/>
    public override async Task<OrchestrationMetadata> WaitForInstanceCompletionAsync(
        string instanceId, bool getInputsAndOutputs = false, CancellationToken cancellation = default)
    {
        Check.NotEntity(this.options.EnableEntitySupport, instanceId);

        this.logger.WaitingForInstanceCompletion(instanceId, getInputsAndOutputs);

        P.GetInstanceRequest request = new()
        {
            InstanceId = instanceId,
            GetInputsAndOutputs = getInputsAndOutputs,
        };

        while (!cancellation.IsCancellationRequested)
        {
            try
            {
                P.GetInstanceResponse response = await this.sidecarClient.WaitForInstanceCompletionAsync(
                    request, cancellationToken: cancellation);
                return this.CreateMetadata(response.OrchestrationState, getInputsAndOutputs);
            }
            catch (RpcException e) when (e.StatusCode == StatusCode.Cancelled)
            {
                throw new OperationCanceledException(
                    $"The {nameof(this.WaitForInstanceCompletionAsync)} operation was canceled.", e, cancellation);
            }
            catch (RpcException e) when (e.StatusCode == StatusCode.DeadlineExceeded)
            {
                // Gateway timeout/deadline exceeded can happen before the request is completed. Do nothing and retry.
            }
        }

        // If the operation was cancelled in between requests, we should still throw instead of returning a null value.
        throw new OperationCanceledException($"The {nameof(this.WaitForInstanceCompletionAsync)} operation was canceled.");
    }

    /// <inheritdoc/>
    public override Task<PurgeResult> PurgeInstanceAsync(
        string instanceId, PurgeInstanceOptions? options = null, CancellationToken cancellation = default)
    {
        Check.NotNullOrEmpty(instanceId);
        bool recursive = options?.Recursive ?? false;
        this.logger.PurgingInstanceMetadata(instanceId);

        P.PurgeInstancesRequest request = new()
        {
            InstanceId = instanceId,
            Recursive = recursive,
            IsOrchestration = !this.options.EnableEntitySupport || instanceId[0] != '@',
        };
        return this.PurgeInstancesCoreAsync(request, cancellation);
    }

    /// <inheritdoc/>
    public override Task<PurgeResult> PurgeAllInstancesAsync(
        PurgeInstancesFilter filter, PurgeInstanceOptions? options = null, CancellationToken cancellation = default)
    {
        bool recursive = options?.Recursive ?? false;
        this.logger.PurgingInstances(filter);
        P.PurgeInstancesRequest request = new()
        {
            PurgeInstanceFilter = new()
            {
                CreatedTimeFrom = filter?.CreatedFrom.ToTimestamp(),
                CreatedTimeTo = filter?.CreatedTo.ToTimestamp(),
            },
            Recursive = recursive,
        };

        if (filter?.Statuses is not null)
        {
            request.PurgeInstanceFilter.RuntimeStatus.AddRange(filter.Statuses.Select(x => x.ToGrpcStatus()));
        }

        if (filter?.Timeout is not null)
        {
            request.PurgeInstanceFilter.Timeout = Google.Protobuf.WellKnownTypes.Duration.FromTimeSpan(filter.Timeout.Value);
        }

        return this.PurgeInstancesCoreAsync(request, cancellation);
    }

    /// <inheritdoc/>
    // Whether or not this method throws a <see cref="InvalidOperationException"/> or terminates the existing instance
    // when <paramref name="restartWithNewInstanceId"/> is <c>false</c> and the existing instance is not in a terminal state
    // depends on the server-side implementation.
    public override async Task<string> RestartAsync(
        string instanceId,
        bool restartWithNewInstanceId = false,
        CancellationToken cancellation = default)
    {
        Check.NotNullOrEmpty(instanceId);
        Check.NotEntity(this.options.EnableEntitySupport, instanceId);

        P.RestartInstanceRequest request = new P.RestartInstanceRequest
        {
            InstanceId = instanceId,
            RestartWithNewInstanceId = restartWithNewInstanceId,
        };

        try
        {
            P.RestartInstanceResponse result = await this.sidecarClient.RestartInstanceAsync(
                request, cancellationToken: cancellation);
            return result.InstanceId;
        }
        catch (RpcException e) when (e.StatusCode == StatusCode.NotFound)
        {
            throw new ArgumentException($"An orchestration with the instanceId {instanceId} was not found.", e);
        }
        catch (RpcException e) when (e.StatusCode == StatusCode.FailedPrecondition)
        {
            throw new InvalidOperationException($"An orchestration with the instanceId {instanceId} cannot be restarted.", e);
        }
        catch (RpcException e) when (e.StatusCode == StatusCode.Cancelled)
        {
            throw new OperationCanceledException(
                $"The {nameof(this.RestartAsync)} operation was canceled.", e, cancellation);
        }
    }

    /// <inheritdoc/>
    public override async Task RewindInstanceAsync(
        string instanceId,
        string reason,
        CancellationToken cancellation = default)
    {
        Check.NotNullOrEmpty(instanceId);
        Check.NotEntity(this.options.EnableEntitySupport, instanceId);

        P.RewindInstanceRequest request = new P.RewindInstanceRequest
        {
            InstanceId = instanceId,
            Reason = reason,
        };
        try
        {
            await this.sidecarClient.RewindInstanceAsync(request, cancellationToken: cancellation);
        }
        catch (RpcException e) when (e.StatusCode == StatusCode.NotFound)
        {
            throw new ArgumentException($"An orchestration with the instanceId {instanceId} was not found.", e);
        }
        catch (RpcException e) when (e.StatusCode == StatusCode.FailedPrecondition)
        {
            throw new InvalidOperationException(e.Status.Detail);
        }
        catch (RpcException e) when (e.StatusCode == StatusCode.Unimplemented)
        {
            throw new NotImplementedException(e.Status.Detail);
        }
        catch (RpcException e) when (e.StatusCode == StatusCode.Cancelled)
        {
            throw new OperationCanceledException(
                $"The {nameof(this.RewindInstanceAsync)} operation was canceled.", e, cancellation);
        }
    }

    /// <inheritdoc/>
    public override async Task<IList<HistoryEvent>> GetOrchestrationHistoryAsync(
        string instanceId,
        CancellationToken cancellation = default)
    {
        Check.NotNullOrEmpty(instanceId);
        Check.NotEntity(this.options.EnableEntitySupport, instanceId);

        P.StreamInstanceHistoryRequest streamRequest = new()
        {
            InstanceId = instanceId,
            ForWorkItemProcessing = false,
        };

        try
        {
            using AsyncServerStreamingCall<P.HistoryChunk> streamResponse =
                this.sidecarClient.StreamInstanceHistory(streamRequest, cancellationToken: cancellation);

            Microsoft.DurableTask.ProtoUtils.EntityConversionState conversionState = new(insertMissingEntityUnlocks: false);
            List<HistoryEvent> pastEvents = [];
            while (await streamResponse.ResponseStream.MoveNext(cancellation))
            {
                pastEvents.AddRange(streamResponse.ResponseStream.Current.Events.Select(conversionState.ConvertFromProto));
            }

            return pastEvents;
        }
        catch (RpcException e) when (e.StatusCode == StatusCode.NotFound)
        {
            throw new ArgumentException($"An orchestration with the instanceId {instanceId} was not found.", e);
        }
        catch (RpcException e) when (e.StatusCode == StatusCode.Cancelled)
        {
            throw new OperationCanceledException(
                $"The {nameof(this.GetOrchestrationHistoryAsync)} operation was canceled.", e, cancellation);
        }
        catch (RpcException e) when (e.StatusCode == StatusCode.Internal)
        {
            throw new InvalidOperationException(
                $"An error occurred while retrieving the history for orchestration with instanceId {instanceId}.", e);
        }
    }

    static AsyncDisposable GetCallInvoker(GrpcDurableTaskClientOptions options, out CallInvoker callInvoker)
    {
        if (options.Channel is GrpcChannel c)
        {
            callInvoker = c.CreateCallInvoker();
            return default;
        }

        if (options.CallInvoker is CallInvoker invoker)
        {
            callInvoker = invoker;
            return default;
        }

        c = GetChannel(options.Address);
        callInvoker = c.CreateCallInvoker();
        return new AsyncDisposable(() => new(c.ShutdownAsync()));
    }

#if NET6_0_OR_GREATER
    static GrpcChannel GetChannel(string? address)
    {
        if (string.IsNullOrEmpty(address))
        {
            address = "http://localhost:4001";
        }

        return GrpcChannel.ForAddress(address);
    }
#endif

#if NETSTANDARD2_0
    static GrpcChannel GetChannel(string? address)
    {
        if (string.IsNullOrEmpty(address))
        {
            address = "localhost:4001";
        }

        return new(address, ChannelCredentials.Insecure);
    }
#endif

    async Task<PurgeResult> PurgeInstancesCoreAsync(
        P.PurgeInstancesRequest request, CancellationToken cancellation = default)
    {
        try
        {
            P.PurgeInstancesResponse response = await this.sidecarClient.PurgeInstancesAsync(
                request, cancellationToken: cancellation);
            return new PurgeResult(response.DeletedInstanceCount, response.IsComplete);
        }
        catch (RpcException e) when (e.StatusCode == StatusCode.Cancelled)
        {
            throw new OperationCanceledException(
                $"The {nameof(this.PurgeAllInstancesAsync)} operation was canceled.", e, cancellation);
        }
        catch (RpcException e) when (e.StatusCode == StatusCode.FailedPrecondition)
        {
            throw new InvalidOperationException(e.Status.Detail);
        }
        catch (RpcException e) when (e.StatusCode == StatusCode.Unimplemented)
        {
            throw new NotImplementedException(e.Status.Detail);
        }
    }

    OrchestrationMetadata CreateMetadata(P.OrchestrationState state, bool includeInputsAndOutputs)
    {
        OrchestrationMetadata metadata = new OrchestrationMetadata(state.Name, state.InstanceId)
        {
            CreatedAt = state.CreatedTimestamp.ToDateTimeOffset(),
            LastUpdatedAt = state.LastUpdatedTimestamp.ToDateTimeOffset(),
            RuntimeStatus = (OrchestrationRuntimeStatus)state.OrchestrationStatus,
            SerializedInput = state.Input,
            SerializedOutput = state.Output,
            SerializedCustomStatus = state.CustomStatus,
            FailureDetails = state.FailureDetails.ToTaskFailureDetails(),
            DataConverter = includeInputsAndOutputs ? this.DataConverter : null,
            Tags = new Dictionary<string, string>(state.Tags),
        };

        return metadata;
    }
}