// Copyright (c) Microsoft. All rights reserved.
using System.Diagnostics.CodeAnalysis;
using System.Text.Json;
using Microsoft.Agents.AI.Workflows;
using Microsoft.DurableTask;
using Microsoft.Extensions.Logging;
#pragma warning disable CA1848 // Use LoggerMessage delegates
#pragma warning disable CA1873 // Expensive evaluation
namespace Microsoft.Agents.AI.DurableTask;
///
/// Represents the custom status set when the orchestration is waiting for an external event.
///
/// The name of the event being waited for (the RequestPort ID).
/// The serialized input data that was passed to the RequestPort.
/// The full type name of the request type.
/// The full type name of the expected response type.
///
///
/// This status is set when a workflow reaches a human-in-the-loop point (RequestPort executor).
/// External actors can query this status to:
///
///
/// Discover which event name to raise
/// See the input/context that prompted the request
/// Understand the expected request and response types for proper serialization
///
///
internal sealed record PendingExternalEventStatus(
string EventName,
string Input,
string RequestType,
string ResponseType);
///
/// Represents the complete custom status for a durable workflow orchestration.
///
///
///
/// This status object is serialized and stored as the orchestration's custom status,
/// making it visible in the Durable Task dashboard and queryable via the client API.
///
///
/// It serves two primary purposes:
///
///
///
/// Event Accumulation: Collects workflow events (yields, halts, etc.) emitted
/// by executors during the workflow run for final result determination.
///
///
/// HITL Status: Indicates when the workflow is waiting for external input,
/// including the event name to raise and context about the request.
///
///
///
internal sealed class DurableWorkflowCustomStatus
{
///
/// Gets or sets the pending external event status when waiting for HITL input.
///
public PendingExternalEventStatus? PendingEvent { get; set; }
///
/// Gets or sets the list of serialized workflow events emitted by executors.
///
public List Events { get; set; } = [];
}
///
/// Core workflow runner that executes workflow orchestrations using Durable Tasks.
/// This class contains the core workflow execution logic independent of the hosting environment.
///
///
///
/// Architecture Overview:
///
///
/// The DurableWorkflowRunner implements a message-driven execution model that naturally supports
/// both directed acyclic graphs (DAGs) and cyclic workflows. Each executor in the workflow is
/// treated as a message processor that receives input, executes logic, and routes output to successors.
///
///
/// Execution Flow:
///
///
///
///
/// Initialization: The workflow starts by queueing the initial input to the start executor.
///
///
///
///
/// Superstep Processing: In each superstep, all executors with pending messages are processed.
/// This continues until no messages remain or a halt is requested.
///
///
///
///
/// Activity Execution: Each executor runs as a Durable Task activity. The activity receives
/// wrapped input containing the message, type information, and shared workflow state.
///
///
///
///
/// Result Unwrapping: Activity results are unwrapped to extract the actual result,
/// state updates, workflow events, and any messages sent via SendMessageAsync.
///
///
///
///
/// Message Routing: Results are routed to successor executors based on workflow edges.
/// Edge conditions are evaluated to determine if a message should be forwarded.
///
///
///
///
/// Key Data Structures:
///
///
///
///
/// Message Queues: Each executor has a queue of pending messages (input + type information).
///
///
///
///
/// Shared State: A dictionary of scope-prefixed key-value pairs shared across executors.
///
///
///
///
/// Custom Status: Tracks workflow events and pending external event info for observability.
///
///
///
///
/// Executor Types:
///
///
///
///
/// Regular Executors: Run as Durable Task activities. Can return values or use SendMessageAsync.
///
///
///
///
/// Agent Executors: AI agents that run via Durable Entities for stateful conversations.
///
///
///
///
/// Request Port Executors: Human-in-the-loop points that wait for external events.
///
///
///
///
/// Input/Output Flow:
///
///
/// ┌─────────────────────────────────────────────────────────────────────────────────┐
/// │ Orchestrator (DurableWorkflowRunner) │
/// │ │
/// │ ┌─────────────┐ ActivityInputWithState ┌─────────────────────────────┐ │
/// │ │ Message │ ──────────────────────────▶ │ Activity │ │
/// │ │ Queue │ {Input, InputTypeName, │ (ExecuteActivityAsync) │ │
/// │ │ │ State} │ │ │
/// │ │ │ │ - Deserialize input │ │
/// │ │ │ │ - Execute executor logic │ │
/// │ │ │ │ - Collect state updates │ │
/// │ │ │ DurableActivityOutput │ - Collect events │ │
/// │ │ │ ◀────────────────────────── │ - Collect sent messages │ │
/// │ └─────────────┘ {Result, StateUpdates, └─────────────────────────────┘ │
/// │ │ ClearedScopes, Events, │
/// │ │ SentMessages} │
/// │ ▼ │
/// │ ┌─────────────┐ │
/// │ │ Route to │ ── Evaluate edge conditions ──▶ Queue to successor executors │
/// │ │ Successors │ │
/// │ └─────────────┘ │
/// └─────────────────────────────────────────────────────────────────────────────────┘
///
///
internal class DurableWorkflowRunner
{
///
/// Initializes a new instance of the class.
///
/// The logger instance.
/// The durable options containing workflow configurations.
public DurableWorkflowRunner(ILogger logger, DurableOptions durableOptions)
{
ArgumentNullException.ThrowIfNull(logger);
ArgumentNullException.ThrowIfNull(durableOptions);
this.Logger = logger;
this.Options = durableOptions.Workflows;
}
///
/// Gets the workflow options.
///
protected DurableWorkflowOptions Options { get; }
///
/// Gets the logger instance.
///
protected ILogger Logger { get; }
///
/// Runs a workflow orchestration.
///
/// The task orchestration context.
/// The workflow run input containing workflow name and input.
/// The replay-safe logger for orchestration logging.
/// The result of the workflow execution.
/// Thrown when the specified workflow is not found.
public async Task RunWorkflowOrchestrationAsync(
TaskOrchestrationContext context,
string input,
ILogger logger)
{
ArgumentNullException.ThrowIfNull(context);
ArgumentNullException.ThrowIfNull(input);
string orchestrationName = context.Name;
string workflowName = WorkflowNamingHelper.ToWorkflowName(orchestrationName);
if (!this.Options.Workflows.TryGetValue(workflowName, out Workflow? workflow))
{
throw new InvalidOperationException($"Workflow '{workflowName}' not found.");
}
logger.LogRunningWorkflow(workflow.Name);
return await this.ExecuteWorkflowAsync(context, workflow, input, logger).ConfigureAwait(true);
}
///
/// Parses the executor name from an activity function name.
///
/// The activity function name.
/// The extracted executor name.
protected static string ParseExecutorName(string activityFunctionName)
{
if (!activityFunctionName.StartsWith(WorkflowNamingHelper.OrchestrationFunctionPrefix, StringComparison.Ordinal))
{
throw new InvalidOperationException(
$"Activity function name '{activityFunctionName}' does not start with '{WorkflowNamingHelper.OrchestrationFunctionPrefix}' prefix.");
}
string executorName = activityFunctionName[WorkflowNamingHelper.OrchestrationFunctionPrefix.Length..];
if (string.IsNullOrEmpty(executorName))
{
throw new InvalidOperationException(
$"Activity function name '{activityFunctionName}' is not in the expected format '{WorkflowNamingHelper.OrchestrationFunctionPrefix}{{executorName}}'.");
}
return executorName;
}
///
/// Serializes a list of strings to JSON using source-generated serialization.
///
protected static string SerializeToJson(List values)
{
return JsonSerializer.Serialize(values, DurableWorkflowJsonContext.Default.ListString);
}
///
/// Serializes a result object to JSON or string.
///
[UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Serializing workflow types registered at startup.")]
[UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Serializing workflow types registered at startup.")]
protected static string SerializeResult(object? result)
{
if (result is null)
{
return string.Empty;
}
if (result is string str)
{
return str;
}
Type resultType = result.GetType();
if (resultType.IsPrimitive || resultType == typeof(decimal))
{
return result.ToString() ?? string.Empty;
}
return JsonSerializer.Serialize(result, resultType);
}
///
/// Deserializes input from JSON to the target type.
///
[UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow types registered at startup.")]
[UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing workflow types registered at startup.")]
protected static object DeserializeInput(string input, Type targetType)
{
if (targetType == typeof(string))
{
return input;
}
string json = input;
// Check for double-serialized strings (e.g., "\"actual json\"")
// A properly double-serialized string starts with \" and ends with \"
if (input.Length > 2 && input[0] == '"' && input[^1] == '"' && input[1] == '\\')
{
string? innerJson = JsonSerializer.Deserialize(input);
if (innerJson is not null)
{
json = innerJson;
}
}
return JsonSerializer.Deserialize(json, targetType)
?? throw new InvalidOperationException($"Failed to deserialize input to type '{targetType.Name}'.");
}
///
/// Executes a workflow by building an execution plan and delegating to message-driven execution.
///
/// The Durable Task orchestration context.
/// The workflow definition to execute.
/// The initial input string to pass to the start executor.
/// The replay-safe logger for orchestration logging.
/// The final result of the workflow execution.
///
/// This method serves as the entry point for workflow execution after the workflow has been
/// resolved from the orchestration name. It builds a that
/// contains the graph structure (successors, predecessors), edge conditions, and executor
/// output types needed for message routing.
///
private async Task ExecuteWorkflowAsync(
TaskOrchestrationContext context,
Workflow workflow,
string initialInput,
ILogger logger)
{
WorkflowExecutionPlan plan = WorkflowHelper.GetExecutionPlan(workflow);
// Use superstep-based execution for all workflows
// This approach naturally handles both DAGs and cyclic workflows
return await this.RunSuperstepLoopAsync(context, workflow, plan, initialInput, logger).ConfigureAwait(true);
}
///
/// Runs the workflow execution loop using superstep-based processing.
///
///
/// Implements Bulk Synchronous Parallel (BSP) style execution where each superstep
/// processes all pending messages for active executors in parallel. Terminates when
/// no messages remain, halt is requested, or max supersteps reached.
///
[UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow types registered at startup.")]
[UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing workflow types registered at startup.")]
private async Task RunSuperstepLoopAsync(
TaskOrchestrationContext context,
Workflow workflow,
WorkflowExecutionPlan plan,
string initialInput,
ILogger logger)
{
const int MaxSupersteps = 100;
SuperstepState state = new(workflow, plan);
EnqueueMessage(state.MessageQueues, plan.StartExecutorId, initialInput, typeof(string).FullName);
for (int superstep = 1; superstep <= MaxSupersteps; superstep++)
{
List inputs = PrepareExecutorInputs(state, logger);
if (inputs.Count == 0)
{
break;
}
logger.LogDebug("Superstep {Step}: {Count} active executor(s)", superstep, inputs.Count);
// Dispatch all executors in parallel
Task[] tasks = inputs
.Select(e => this.DispatchExecutorAsync(context, e.Info, e.Input, e.InputTypeName, logger, state.CustomStatus, state.SharedState))
.ToArray();
string[] results = await Task.WhenAll(tasks).ConfigureAwait(true);
// Process results and route to successors
string? haltOutput = ProcessSuperstepResults(inputs, results, state, logger);
if (haltOutput is not null)
{
return haltOutput;
}
UpdateCustomStatus(context, state.CustomStatus);
}
return DetermineFinalResult(workflow, state.LastResults, state.CustomStatus);
}
///
/// Holds mutable state for superstep-based workflow execution.
///
private sealed class SuperstepState(Workflow workflow, WorkflowExecutionPlan plan)
{
public WorkflowExecutionPlan Plan { get; } = plan;
public Dictionary ExecutorBindings { get; } = workflow.ReflectExecutors();
public Dictionary> MessageQueues { get; } = [];
public Dictionary LastResults { get; } = [];
public Dictionary SharedState { get; } = [];
public DurableWorkflowCustomStatus CustomStatus { get; } = new();
}
///
/// Represents prepared input for an executor ready for dispatch.
///
private sealed record ExecutorInput(string ExecutorId, string Input, string? InputTypeName, WorkflowExecutorInfo Info);
///
/// Prepares inputs for all active executors, handling Fan-In aggregation.
///
private static List PrepareExecutorInputs(SuperstepState state, ILogger logger)
{
List inputs = [];
foreach ((string executorId, Queue<(string Message, string? InputTypeName)> queue) in state.MessageQueues)
{
if (queue.Count == 0)
{
continue;
}
bool isFanIn = state.Plan.Predecessors.TryGetValue(executorId, out List? predecessors) && predecessors.Count > 1;
(string input, string? inputTypeName) = isFanIn && queue.Count > 1
? AggregateQueueMessages(queue, executorId, logger)
: queue.Dequeue();
inputs.Add(new ExecutorInput(executorId, input, inputTypeName, CreateExecutorInfo(executorId, state.ExecutorBindings)));
}
return inputs;
}
///
/// Aggregates all messages in a queue into a JSON array for Fan-In executors.
///
private static (string Input, string? TypeName) AggregateQueueMessages(
Queue<(string Message, string? InputTypeName)> queue,
string executorId,
ILogger logger)
{
List messages = [];
while (queue.Count > 0)
{
messages.Add(queue.Dequeue().Message);
}
logger.LogDebug("Fan-In executor {ExecutorId}: aggregated {Count} messages", executorId, messages.Count);
return (AggregateMessagesToJsonArray(messages), typeof(string[]).FullName);
}
///
/// Processes results from a superstep, updating state and routing messages.
///
/// The halt output if halt was requested; otherwise null.
private static string? ProcessSuperstepResults(
List inputs,
string[] rawResults,
SuperstepState state,
ILogger logger)
{
for (int i = 0; i < inputs.Count; i++)
{
string executorId = inputs[i].ExecutorId;
(string result, List sentMessages) = UnwrapActivityResult(rawResults[i], state.CustomStatus, state.SharedState);
state.LastResults[executorId] = result;
if (HasHaltBeenRequested(state.CustomStatus, executorId))
{
logger.LogDebug("Halt requested by executor {ExecutorId}", executorId);
return result;
}
RouteExecutorOutput(executorId, result, sentMessages, state, logger);
}
return null;
}
///
/// Routes executor output (explicit messages or return value) to successors.
///
private static void RouteExecutorOutput(
string executorId,
string result,
List sentMessages,
SuperstepState state,
ILogger logger)
{
if (sentMessages.Count > 0)
{
foreach (SentMessageInfo msg in sentMessages)
{
if (!string.IsNullOrEmpty(msg.Message))
{
RouteMessageToSuccessors(executorId, msg.Message, msg.TypeName, state.Plan, state.MessageQueues, logger);
}
}
}
else if (!string.IsNullOrEmpty(result))
{
RouteMessageToSuccessors(executorId, result, state.Plan, state.MessageQueues, logger);
}
}
///
/// Aggregates multiple messages into a JSON array.
///
/// The messages to aggregate.
/// A JSON array string containing all messages.
[UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Serializing string array.")]
[UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Serializing string array.")]
private static string AggregateMessagesToJsonArray(List messages)
{
return JsonSerializer.Serialize(messages);
}
///
/// Enqueues a message to an executor's message queue with type information.
///
/// The dictionary of message queues, keyed by executor ID.
/// The target executor ID to queue the message for.
/// The serialized message content.
/// The full type name of the message, used for deserialization hints.
///
/// Creates a new queue for the executor if one doesn't exist. Messages are processed
/// in FIFO order during each superstep.
///
private static void EnqueueMessage(
Dictionary> queues,
string executorId,
string message,
string? inputTypeName)
{
if (!queues.TryGetValue(executorId, out Queue<(string, string?)>? queue))
{
queue = new Queue<(string, string?)>();
queues[executorId] = queue;
}
queue.Enqueue((message, inputTypeName));
}
///
/// Creates a for the given executor ID.
///
/// The executor ID to look up.
/// The dictionary of executor bindings from the workflow.
/// A containing metadata about the executor.
/// Thrown when the executor ID is not found in bindings.
///
/// This method determines the executor type (agentic, sub-workflow, request port, or regular)
/// and extracts the appropriate metadata for each type.
///
private static WorkflowExecutorInfo CreateExecutorInfo(
string executorId,
Dictionary executorBindings)
{
if (!executorBindings.TryGetValue(executorId, out ExecutorBinding? binding))
{
throw new InvalidOperationException($"Executor '{executorId}' not found in workflow bindings.");
}
bool isAgentic = WorkflowHelper.IsAgentExecutorType(binding.ExecutorType);
RequestPort? requestPort = (binding is RequestPortBinding rpb) ? rpb.Port : null;
Workflow? subWorkflow = (binding is SubworkflowBinding swb) ? swb.WorkflowInstance : null;
return new WorkflowExecutorInfo(executorId, isAgentic, requestPort, subWorkflow);
}
///
/// Checks if the workflow should halt based on halt request events.
///
/// The custom status containing accumulated workflow events.
/// The executor ID that just completed execution.
/// true if a halt was requested; otherwise, false.
///
///
/// Only DurableHaltRequestedEvent triggers a halt. Note that YieldOutputAsync
/// does NOT halt the workflow - it just yields intermediate output that can be used as
/// the final result if no other output is produced.
///
///
/// Events are searched in reverse order (most recent first) for efficiency.
///
///
private static bool HasHaltBeenRequested(
DurableWorkflowCustomStatus customStatus,
string executorId)
{
// Look for explicit halt request events from this executor
for (int i = customStatus.Events.Count - 1; i >= 0; i--)
{
string eventJson = customStatus.Events[i];
// Check for DurableHaltRequestedEvent - this is the ONLY event that should halt
if (eventJson.Contains("DurableHaltRequestedEvent", StringComparison.Ordinal) &&
eventJson.Contains(executorId, StringComparison.Ordinal))
{
return true;
}
}
return false;
}
///
/// Routes a message through edges to successor executors.
///
/// The source executor ID.
/// The serialized message to route.
/// The workflow execution plan.
/// The message queues for each executor.
/// The logger instance.
[UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow types registered at startup.")]
[UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing workflow types registered at startup.")]
private static void RouteMessageToSuccessors(
string sourceId,
string message,
WorkflowExecutionPlan plan,
Dictionary> messageQueues,
ILogger logger)
{
plan.ExecutorOutputTypes.TryGetValue(sourceId, out Type? sourceOutputType);
RouteMessageToSuccessorsCore(sourceId, message, sourceOutputType?.FullName, sourceOutputType, plan, messageQueues, logger);
}
///
/// Routes a message through edges to successor executors, with an explicit type name.
/// Used for messages sent via SendMessageAsync.
///
[UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow types registered at startup.")]
[UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing workflow types registered at startup.")]
[UnconditionalSuppressMessage("Trimming", "IL2057", Justification = "Type resolution for workflow message types.")]
private static void RouteMessageToSuccessors(
string sourceId,
string message,
string? explicitTypeName,
WorkflowExecutionPlan plan,
Dictionary> messageQueues,
ILogger logger)
{
Type? messageType = !string.IsNullOrEmpty(explicitTypeName) ? Type.GetType(explicitTypeName) : null;
string? inputTypeName = explicitTypeName;
if (messageType is null && plan.ExecutorOutputTypes.TryGetValue(sourceId, out Type? sourceOutputType))
{
messageType = sourceOutputType;
inputTypeName = sourceOutputType?.FullName;
}
RouteMessageToSuccessorsCore(sourceId, message, inputTypeName, messageType, plan, messageQueues, logger);
}
///
/// Core implementation for routing messages to successor executors.
///
/// The source executor ID that produced the message.
/// The serialized message content to route.
/// The type name to pass to successors for deserialization.
/// The resolved for edge condition evaluation.
/// The workflow execution plan containing the graph structure.
/// The message queues to enqueue messages into.
/// The logger for debug output.
///
/// For each successor of the source executor, this method:
///
/// Evaluates the edge condition (if any)
/// Enqueues the message if the condition passes or no condition exists
///
///
[UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow types registered at startup.")]
[UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing workflow types registered at startup.")]
private static void RouteMessageToSuccessorsCore(
string sourceId,
string message,
string? inputTypeName,
Type? messageType,
WorkflowExecutionPlan plan,
Dictionary> messageQueues,
ILogger logger)
{
if (!plan.Successors.TryGetValue(sourceId, out List? successors))
{
return;
}
foreach (string sinkId in successors)
{
if (!TryEvaluateEdgeCondition(sourceId, sinkId, message, messageType, plan, logger))
{
continue;
}
logger.LogDebug("Edge {Source} -> {Sink}: routing message", sourceId, sinkId);
EnqueueMessage(messageQueues, sinkId, message, inputTypeName);
}
}
///
/// Evaluates an edge condition if one exists for the given source-sink pair.
///
/// The source executor ID.
/// The target (sink) executor ID.
/// The serialized message to evaluate.
/// The type to deserialize the message to for condition evaluation.
/// The execution plan containing edge conditions.
/// The logger for debug/warning output.
///
/// true if the message should be routed (no condition exists or condition passed);
/// false if the condition returned false or evaluation failed.
///
///
/// Edge conditions are user-defined predicates that filter which messages flow through an edge.
/// If evaluation throws an exception, the edge is skipped (fail-safe behavior) and a warning is logged.
///
[UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow types registered at startup.")]
[UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing workflow types registered at startup.")]
private static bool TryEvaluateEdgeCondition(
string sourceId,
string sinkId,
string message,
Type? messageType,
WorkflowExecutionPlan plan,
ILogger logger)
{
if (!plan.EdgeConditions.TryGetValue((sourceId, sinkId), out Func? condition) || condition is null)
{
return true;
}
try
{
object? messageObj = DeserializeMessageForCondition(message, messageType);
if (!condition(messageObj))
{
logger.LogDebug("Edge {Source} -> {Sink}: condition returned false, skipping", sourceId, sinkId);
return false;
}
return true;
}
catch (Exception ex)
{
logger.LogWarning(ex, "Failed to evaluate condition for edge {Source} -> {Sink}, skipping", sourceId, sinkId);
return false;
}
}
///
/// Determines the final result for a workflow execution.
///
/// The workflow definition.
/// Dictionary of last results from each executor.
/// The custom status containing workflow events.
/// The final workflow result string.
///
///
/// Result priority (highest to lowest):
///
///
/// Most recent YieldOutputAsync call (explicit workflow output)
/// Results from designated output executors (via WithOutputFrom)
/// Last non-empty result from any executor
///
///
/// If multiple output executors are defined, their results are joined with \n---\n.
///
///
[UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing event types.")]
[UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing event types.")]
private static string DetermineFinalResult(
Workflow workflow,
Dictionary lastResults,
DurableWorkflowCustomStatus customStatus)
{
// First, check for yielded outputs from YieldOutputAsync calls (most recent first)
// These take priority as they represent explicit workflow outputs
string? yieldedOutput = GetLastYieldedOutput(customStatus);
if (!string.IsNullOrEmpty(yieldedOutput))
{
return yieldedOutput;
}
HashSet outputExecutors = workflow.ReflectOutputExecutors();
// If specific output executors are defined, use their results
if (outputExecutors.Count > 0)
{
List outputResults = [];
foreach (string outputExecutorId in outputExecutors)
{
if (lastResults.TryGetValue(outputExecutorId, out string? result) && !string.IsNullOrEmpty(result))
{
outputResults.Add(result);
}
}
if (outputResults.Count > 0)
{
return outputResults.Count == 1
? outputResults[0]
: string.Join("\n---\n", outputResults);
}
}
// Otherwise, return the last non-empty result from any executor
return lastResults.Values.LastOrDefault(v => !string.IsNullOrEmpty(v)) ?? string.Empty;
}
///
/// Extracts the most recent yielded output from the custom status events.
///
/// The custom status containing serialized workflow events.
/// The yielded output string, or null if no yield event was found.
///
///
/// Searches for DurableYieldedOutputEvent in reverse order (most recent first).
/// The event structure is:
///
///
/// {
/// "TypeName": "...DurableYieldedOutputEvent...",
/// "Data": "{\"Output\": \"actual output value\"}"
/// }
///
///
/// The output can be either a string or a serialized object (returned as raw JSON).
///
///
[UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing event types.")]
[UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing event types.")]
private static string? GetLastYieldedOutput(DurableWorkflowCustomStatus customStatus)
{
// Look for DurableYieldedOutputEvent in events (most recent first)
for (int i = customStatus.Events.Count - 1; i >= 0; i--)
{
string eventJson = customStatus.Events[i];
if (eventJson.Contains("DurableYieldedOutputEvent", StringComparison.Ordinal))
{
try
{
// Parse the wrapper to get the inner Data
using JsonDocument doc = JsonDocument.Parse(eventJson);
if (doc.RootElement.TryGetProperty("Data", out JsonElement dataElement))
{
string? dataJson = dataElement.GetString();
if (dataJson is not null)
{
using JsonDocument dataDoc = JsonDocument.Parse(dataJson);
if (dataDoc.RootElement.TryGetProperty("Output", out JsonElement outputElement))
{
// The output could be a string or a serialized object
return outputElement.ValueKind == JsonValueKind.String
? outputElement.GetString()
: outputElement.GetRawText();
}
}
}
}
catch (JsonException)
{
// Continue to next event if parsing fails
}
}
}
return null;
}
///
/// Unwraps an activity result, extracting state updates, events, sent messages, and the actual result.
///
/// The raw JSON string returned from the activity.
/// The custom status to append events to.
/// The shared state dictionary to apply updates to.
///
/// A tuple containing the executor's result and any messages sent via SendMessageAsync.
///
///
///
/// Activities return a wrapper containing:
///
///
/// Result: The serialized return value of the executor
/// StateUpdates: Key-value pairs to add/update/remove from shared state
/// ClearedScopes: Scope prefixes to bulk-remove from shared state
/// Events: Workflow events (yield, halt, etc.) to propagate
/// SentMessages: Messages sent via SendMessageAsync
///
///
/// If the raw result is not a valid , it's returned as-is
/// (backward compatibility for activities that return plain values).
///
///
private static (string Result, List SentMessages) UnwrapActivityResult(
string rawResult,
DurableWorkflowCustomStatus customStatus,
Dictionary sharedState)
{
if (string.IsNullOrEmpty(rawResult))
{
return (rawResult, []);
}
try
{
DurableActivityOutput? output = JsonSerializer.Deserialize(rawResult, DurableWorkflowJsonContext.Default.DurableActivityOutput);
if (output is null || !IsValidActivityOutput(output))
{
return (rawResult, []);
}
ApplyClearedScopes(output.ClearedScopes, sharedState);
ApplyStateUpdates(output.StateUpdates, sharedState);
if (output.Events.Count > 0)
{
customStatus.Events.AddRange(output.Events);
}
return (output.Result ?? string.Empty, output.SentMessages);
}
catch (JsonException)
{
return (rawResult, []);
}
}
///
/// Checks if the deserialized output is a valid (not just default values).
///
/// The deserialized output to validate.
/// true if the output has any meaningful content; otherwise, false.
///
/// This check distinguishes actual activity output from arbitrary JSON that happened to
/// deserialize successfully but with all default/empty values.
///
private static bool IsValidActivityOutput(DurableActivityOutput output)
{
return output.Result is not null
|| output.StateUpdates.Count > 0
|| output.ClearedScopes.Count > 0
|| output.Events.Count > 0
|| output.SentMessages.Count > 0;
}
///
/// Clears all state entries matching the specified scopes.
///
/// The list of scope names to clear.
/// The shared state dictionary to modify.
///
///
/// State keys are prefixed with their scope (e.g., myScope:keyName).
/// The special scope __default__ is used for keys without an explicit scope.
///
///
/// Clearing a scope removes all keys that start with {scope}:.
///
///
private static void ApplyClearedScopes(List clearedScopes, Dictionary sharedState)
{
foreach (string clearedScope in clearedScopes)
{
string scopePrefix = clearedScope == "__default__" ? "__default__:" : $"{clearedScope}:";
List keysToRemove = sharedState.Keys
.Where(k => k.StartsWith(scopePrefix, StringComparison.Ordinal))
.ToList();
foreach (string key in keysToRemove)
{
sharedState.Remove(key);
}
}
}
///
/// Applies state updates to the shared state dictionary.
///
/// The updates to apply (key -> value, where null value means delete).
/// The shared state dictionary to modify.
///
/// A null value in the updates dictionary signals that the key should be removed
/// from the shared state, enabling executors to delete state entries.
///
private static void ApplyStateUpdates(Dictionary stateUpdates, Dictionary sharedState)
{
foreach (KeyValuePair update in stateUpdates)
{
if (update.Value is null)
{
sharedState.Remove(update.Key);
}
else
{
sharedState[update.Key] = update.Value;
}
}
}
///
/// Updates the orchestration custom status with current events and pending event info.
///
/// The orchestration context to set the status on.
/// The custom status object containing events and pending event info.
///
/// The custom status is visible in the Durable Task dashboard and can be queried via the
/// Durable Task client. It includes:
///
/// Accumulated workflow events from all executors
/// Pending external event info when waiting for human-in-the-loop input
///
/// Only updates if there's meaningful content to report.
///
private static void UpdateCustomStatus(TaskOrchestrationContext context, DurableWorkflowCustomStatus customStatus)
{
// Only update if there are events or a pending event
if (customStatus.Events.Count > 0 || customStatus.PendingEvent is not null)
{
context.SetCustomStatus(customStatus);
}
}
///
/// Deserializes a JSON message string into an object for edge condition evaluation.
///
/// The JSON string to deserialize.
/// The target type to deserialize to, or null to deserialize as .
///
/// The deserialized object, or the original string if deserialization fails (graceful fallback).
///
///
/// This method is used to prepare the message for edge condition predicates.
/// If the JSON is invalid, it returns the raw string to allow conditions to handle string inputs.
///
[UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow types registered at startup.")]
[UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing workflow types registered at startup.")]
private static object? DeserializeMessageForCondition(string json, Type? targetType)
{
if (string.IsNullOrEmpty(json))
{
return null;
}
try
{
if (targetType is null)
{
return JsonSerializer.Deserialize