From fcd7aa0b77942afa088d36763d3dc3b63f53c5d0 Mon Sep 17 00:00:00 2001 From: Shyju Krishnankutty Date: Tue, 20 Jan 2026 16:22:19 -0800 Subject: [PATCH] Move relevant stuff to DurableTask project from Hosting.AzureFunctions --- .../01_ExecutorsAndEdges.csproj | 22 ++ .../01_ExecutorsAndEdges/Program.cs | 63 ++++ .../DurableExecutorContext.cs | 6 +- .../DurableWorkflowRunRequest.cs} | 4 +- .../DurableWorkflowRunner.cs | 320 ++++++++++++++++++ .../Microsoft.Agents.AI.DurableTask/Logs.cs | 30 ++ .../Microsoft.Agents.AI.DurableTask.csproj | 1 + .../WorkflowHelper.cs | 15 +- .../WorkflowSharedStateEntity.cs | 10 +- .../BuiltInFunctions.cs | 6 +- .../CoreAgentConfigurationExtensions.cs | 2 +- .../DurableOptionsExtensions.cs | 2 +- ...ableWorkflowFunctionMetadataTransformer.cs | 14 +- .../DurableWorkflowRunner.cs | 285 +--------------- .../DurableWorkflowRunnerLogs.cs | 38 --- .../MinimalActivityContext.cs | 0 16 files changed, 469 insertions(+), 349 deletions(-) create mode 100644 dotnet/samples/DurableWorkflows/01_ExecutorsAndEdges/01_ExecutorsAndEdges.csproj create mode 100644 dotnet/samples/DurableWorkflows/01_ExecutorsAndEdges/Program.cs rename dotnet/src/{Microsoft.Agents.AI.Hosting.AzureFunctions => Microsoft.Agents.AI.DurableTask}/DurableExecutorContext.cs (98%) rename dotnet/src/{Microsoft.Agents.AI.Hosting.AzureFunctions/DuableWorkflowRunRequest.cs => Microsoft.Agents.AI.DurableTask/DurableWorkflowRunRequest.cs} (84%) create mode 100644 dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunner.cs rename dotnet/src/{Microsoft.Agents.AI.Hosting.AzureFunctions => Microsoft.Agents.AI.DurableTask}/WorkflowHelper.cs (94%) rename dotnet/src/{Microsoft.Agents.AI.Hosting.AzureFunctions => Microsoft.Agents.AI.DurableTask}/WorkflowSharedStateEntity.cs (94%) delete mode 100644 dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowRunnerLogs.cs delete mode 100644 dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/MinimalActivityContext.cs diff --git a/dotnet/samples/DurableWorkflows/01_ExecutorsAndEdges/01_ExecutorsAndEdges.csproj b/dotnet/samples/DurableWorkflows/01_ExecutorsAndEdges/01_ExecutorsAndEdges.csproj new file mode 100644 index 0000000000..cdbf5b5805 --- /dev/null +++ b/dotnet/samples/DurableWorkflows/01_ExecutorsAndEdges/01_ExecutorsAndEdges.csproj @@ -0,0 +1,22 @@ + + + + Exe + net10.0 + + enable + enable + + + + + + + + + + + + + + diff --git a/dotnet/samples/DurableWorkflows/01_ExecutorsAndEdges/Program.cs b/dotnet/samples/DurableWorkflows/01_ExecutorsAndEdges/Program.cs new file mode 100644 index 0000000000..af1dcb50d9 --- /dev/null +++ b/dotnet/samples/DurableWorkflows/01_ExecutorsAndEdges/Program.cs @@ -0,0 +1,63 @@ +// Copyright (c) Microsoft. All rights reserved. + +using Microsoft.Agents.AI.Workflows; + +namespace WorkflowExecutorsAndEdgesSample; + +/// +/// This sample introduces the concepts of executors and edges in a workflow. +/// +/// Workflows are built from executors (processing units) connected by edges (data flow paths). +/// In this example, we create a simple text processing pipeline that: +/// 1. Takes input text and converts it to uppercase using an UppercaseExecutor +/// 2. Takes the uppercase text and reverses it using a ReverseTextExecutor +/// +/// The executors are connected sequentially, so data flows from one to the next in order. +/// For input "Hello, World!", the workflow produces "!DLROW ,OLLEH". +/// +public static class Program +{ + private static async Task Main() + { + // Create the executors + Func uppercaseFunc = s => s.ToUpperInvariant(); + var uppercase = uppercaseFunc.BindAsExecutor("UppercaseExecutor"); + + ReverseTextExecutor reverse = new(); + + // Build the workflow by connecting executors sequentially + WorkflowBuilder builder = new(uppercase); + builder.AddEdge(uppercase, reverse).WithOutputFrom(reverse); + var workflow = builder.Build(); + + // Execute the workflow with input data + await using Run run = await InProcessExecution.RunAsync(workflow, "Hello, World!"); + foreach (WorkflowEvent evt in run.NewEvents) + { + if (evt is ExecutorCompletedEvent executorComplete) + { + Console.WriteLine($"{executorComplete.ExecutorId}: {executorComplete.Data}"); + } + } + } +} + +/// +/// Second executor: reverses the input text and completes the workflow. +/// +internal sealed class ReverseTextExecutor() : Executor("ReverseTextExecutor") +{ + /// + /// Processes the input message by reversing the text. + /// + /// The input text to reverse + /// Workflow context for accessing workflow services and adding events + /// The to monitor for cancellation requests. + /// The default is . + /// The input text reversed + public override ValueTask HandleAsync(string message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + // Because we do not suppress it, the returned result will be yielded as an output from this executor. + return ValueTask.FromResult(string.Concat(message.Reverse())); + } +} diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableExecutorContext.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableExecutorContext.cs similarity index 98% rename from dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableExecutorContext.cs rename to dotnet/src/Microsoft.Agents.AI.DurableTask/DurableExecutorContext.cs index e247ed2cd0..325145682e 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableExecutorContext.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableExecutorContext.cs @@ -7,10 +7,10 @@ using Microsoft.DurableTask.Client; using Microsoft.DurableTask.Client.Entities; using Microsoft.DurableTask.Entities; -namespace Microsoft.Agents.AI.Hosting.AzureFunctions; +namespace Microsoft.Agents.AI.DurableTask; /// -/// An implementation of for workflow executors running as Azure Functions activities. +/// An implementation of for workflow executors running as durable activities. /// Provides durable state management using Durable Entities. State is scoped to the orchestration instance /// and shared between executors running on potentially different compute instances. /// @@ -21,7 +21,7 @@ namespace Microsoft.Agents.AI.Hosting.AzureFunctions; /// [RequiresUnreferencedCode("State serialization uses reflection-based JSON serialization.")] [RequiresDynamicCode("State serialization uses reflection-based JSON serialization.")] -internal sealed class DurableExecutorContext : IWorkflowContext +public sealed class DurableExecutorContext : IWorkflowContext { private readonly string _instanceId; private readonly DurableTaskClient _client; diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DuableWorkflowRunRequest.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunRequest.cs similarity index 84% rename from dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DuableWorkflowRunRequest.cs rename to dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunRequest.cs index 9b3fa7db5e..037ea8291c 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DuableWorkflowRunRequest.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunRequest.cs @@ -2,12 +2,12 @@ using System.Text.Json.Serialization; -namespace Microsoft.Agents.AI.Hosting.AzureFunctions; +namespace Microsoft.Agents.AI.DurableTask; /// /// Represents a request to run a durable workflow. /// -internal sealed class DuableWorkflowRunRequest +public sealed class DurableWorkflowRunRequest { /// /// Gets or sets the name of the workflow to execute. diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunner.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunner.cs new file mode 100644 index 0000000000..becdf348fa --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunner.cs @@ -0,0 +1,320 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Diagnostics.CodeAnalysis; +using System.Text.Json; +using Microsoft.Agents.AI.Workflows; +using Microsoft.DurableTask; +using Microsoft.DurableTask.Entities; +using Microsoft.Extensions.Logging; + +namespace Microsoft.Agents.AI.DurableTask; + +/// +/// Core workflow runner that executes workflow orchestrations using Durable Tasks. +/// This class contains the core workflow execution logic independent of the hosting environment. +/// +public 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 request containing workflow name and input. + /// The replay-safe logger for orchestration logging. + /// A list containing the workflow execution result. + public async Task> RunWorkflowOrchestrationAsync( + TaskOrchestrationContext context, + DurableWorkflowRunRequest request, + ILogger logger) + { + ArgumentNullException.ThrowIfNull(context); + ArgumentNullException.ThrowIfNull(request); + + if (!this.Options.Workflows.TryGetValue(request.WorkflowName, out Workflow? workflow)) + { + throw new InvalidOperationException($"Workflow '{request.WorkflowName}' not found."); + } + + logger.LogRunningWorkflow(workflow.Name); + + string result = await this.ExecuteWorkflowLevelsAsync(context, workflow, request.Input, logger).ConfigureAwait(true); + + await CleanupWorkflowStateAsync(context).ConfigureAwait(true); + + return [result]; + } + + /// + /// Cleans up the workflow state entity by signaling it to delete itself. + /// + private static async Task CleanupWorkflowStateAsync(TaskOrchestrationContext context) + { + EntityInstanceId stateEntityId = new(WorkflowSharedStateEntity.EntityName, context.InstanceId); + + // Call the entity's Delete method to clean up state + // Using CallEntityAsync ensures the deletion completes before the orchestration finishes + await context.Entities.CallEntityAsync(stateEntityId, nameof(WorkflowSharedStateEntity.Delete)).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) + { + const string Prefix = "dafx-"; + + if (!activityFunctionName.StartsWith(Prefix, StringComparison.Ordinal)) + { + throw new InvalidOperationException( + $"Activity function name '{activityFunctionName}' does not start with '{Prefix}' prefix."); + } + + string executorName = activityFunctionName[Prefix.Length..]; + + if (string.IsNullOrEmpty(executorName)) + { + throw new InvalidOperationException( + $"Activity function name '{activityFunctionName}' is not in the expected format '{Prefix}{{executorName}}'."); + } + + return executorName; + } + + /// + /// Gets the base name from an executor ID by removing any GUID suffix. + /// + /// The executor ID. + /// The base name without the GUID suffix. + protected static string GetBaseName(string executorId) + { + int underscoreIndex = executorId.IndexOf('_'); + return underscoreIndex > 0 ? executorId[..underscoreIndex] : executorId; + } + + /// + /// Serializes a list of strings to JSON. + /// + [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Serializing known types.")] + [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Serializing known types.")] + protected static string SerializeToJson(List values) + { + return JsonSerializer.Serialize(values); + } + + /// + /// 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; + if (input.StartsWith('"') && input.EndsWith('"')) + { + try + { + string? innerJson = JsonSerializer.Deserialize(input); + if (innerJson is not null) + { + json = innerJson; + } + } + catch (JsonException) + { + // Not double-serialized, use original + } + } + + return JsonSerializer.Deserialize(json, targetType) + ?? throw new InvalidOperationException($"Failed to deserialize input to type '{targetType.Name}'."); + } + + private async Task ExecuteWorkflowLevelsAsync( + TaskOrchestrationContext context, + Workflow workflow, + string initialInput, + ILogger logger) + { + WorkflowExecutionPlan plan = WorkflowHelper.GetExecutionPlan(workflow); + Dictionary results = []; + + foreach (WorkflowExecutionLevel level in plan.Levels) + { + if (level.Executors.Count == 1) + { + WorkflowExecutorInfo executorInfo = level.Executors[0]; + string input = GetExecutorInput(executorInfo.ExecutorId, initialInput, results, plan); + results[executorInfo.ExecutorId] = await this.ExecuteExecutorAsync(context, executorInfo, input, logger).ConfigureAwait(true); + } + else + { + List> tasks = []; + foreach (WorkflowExecutorInfo executorInfo in level.Executors) + { + string input = GetExecutorInput(executorInfo.ExecutorId, initialInput, results, plan); + tasks.Add(this.ExecuteExecutorWithIdAsync(context, executorInfo, input, logger)); + } + + foreach ((string id, string result) in await Task.WhenAll(tasks).ConfigureAwait(true)) + { + results[id] = result; + } + } + } + + return GetFinalResult(plan, results); + } + + private async Task<(string Id, string Result)> ExecuteExecutorWithIdAsync( + TaskOrchestrationContext context, + WorkflowExecutorInfo executorInfo, + string input, + ILogger logger) + { + string result = await this.ExecuteExecutorAsync(context, executorInfo, input, logger).ConfigureAwait(true); + return (executorInfo.ExecutorId, result); + } + + private async Task ExecuteExecutorAsync( + TaskOrchestrationContext context, + WorkflowExecutorInfo executorInfo, + string input, + ILogger logger) + { + if (!executorInfo.IsAgenticExecutor) + { + string triggerName = $"dafx-{GetBaseName(executorInfo.ExecutorId)}"; + return await context.CallActivityAsync(triggerName, input).ConfigureAwait(true); + } + + return await ExecuteAgentAsync(context, executorInfo, input, logger).ConfigureAwait(true); + } + + private static async Task ExecuteAgentAsync( + TaskOrchestrationContext context, + WorkflowExecutorInfo executorInfo, + string input, + ILogger logger) + { + string agentName = GetBaseName(executorInfo.ExecutorId); + DurableAIAgent agent = context.GetAgent(agentName); + + if (agent is null) + { + logger.LogWarning("Agent '{AgentName}' not found", agentName); + return $"Agent '{agentName}' not found"; + } + + AgentThread thread = agent.GetNewThread(); + AgentRunResponse response = await agent.RunAsync(input, thread).ConfigureAwait(true); + return response.Text; + } + + private static string GetExecutorInput( + string executorId, + string initialInput, + Dictionary results, + WorkflowExecutionPlan plan) + { + List predecessors = plan.Predecessors[executorId]; + + if (predecessors.Count == 0) + { + return initialInput; + } + + if (predecessors.Count == 1) + { + return results.TryGetValue(predecessors[0], out string? result) ? result : initialInput; + } + + List aggregated = []; + foreach (string predecessorId in predecessors) + { + if (results.TryGetValue(predecessorId, out string? result)) + { + aggregated.Add(result); + } + } + + return SerializeToJson(aggregated); + } + + private static string GetFinalResult(WorkflowExecutionPlan plan, Dictionary results) + { + WorkflowExecutionLevel lastLevel = plan.Levels[^1]; + + if (lastLevel.Executors.Count == 1) + { + return results[lastLevel.Executors[0].ExecutorId]; + } + + List finalResults = []; + foreach (WorkflowExecutorInfo executor in lastLevel.Executors) + { + if (results.TryGetValue(executor.ExecutorId, out string? result)) + { + finalResults.Add(result); + } + } + + return string.Join("\n---\n", finalResults); + } +} diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/Logs.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/Logs.cs index ba310441df..86794199d4 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/Logs.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/Logs.cs @@ -100,4 +100,34 @@ internal static partial class Logs public static partial void LogTTLExpirationTimeCleared( this ILogger logger, AgentSessionId sessionId); + + [LoggerMessage( + EventId = 12, + Level = LogLevel.Information, + Message = "Attempting to run workflow: {WorkflowName}")] + public static partial void LogAttemptingToRunWorkflow(this ILogger logger, string workflowName); + + [LoggerMessage( + EventId = 13, + Level = LogLevel.Information, + Message = "Running workflow: {WorkflowName}")] + public static partial void LogRunningWorkflow(this ILogger logger, string? workflowName); + + [LoggerMessage( + EventId = 14, + Level = LogLevel.Information, + Message = "Attempting to execute activity in workflow '{WorkflowName}' for executor '{ExecutorName}'")] + public static partial void LogAttemptingToExecuteActivity(this ILogger logger, string workflowName, string executorName); + + [LoggerMessage( + EventId = 15, + Level = LogLevel.Information, + Message = "Executing activity for executor '{ExecutorId}' of type '{ExecutorType}'")] + public static partial void LogExecutingActivity(this ILogger logger, string executorId, string executorType); + + [LoggerMessage( + EventId = 16, + Level = LogLevel.Information, + Message = "Activity executed for executor '{ExecutorId}' with result: {Result}")] + public static partial void LogActivityExecuted(this ILogger logger, string executorId, string result); } diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/Microsoft.Agents.AI.DurableTask.csproj b/dotnet/src/Microsoft.Agents.AI.DurableTask/Microsoft.Agents.AI.DurableTask.csproj index 782004ec6e..e970194fa9 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/Microsoft.Agents.AI.DurableTask.csproj +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/Microsoft.Agents.AI.DurableTask.csproj @@ -21,6 +21,7 @@ + diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/WorkflowHelper.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowHelper.cs similarity index 94% rename from dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/WorkflowHelper.cs rename to dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowHelper.cs index 2f89291675..600761fda5 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/WorkflowHelper.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowHelper.cs @@ -3,14 +3,14 @@ using Microsoft.Agents.AI.Workflows; using Microsoft.Agents.AI.Workflows.Checkpointing; -namespace Microsoft.Agents.AI.Hosting.AzureFunctions; +namespace Microsoft.Agents.AI.DurableTask; /// /// Represents an executor in the workflow with its metadata. /// /// The unique identifier of the executor. /// Indicates whether this executor is an agentic executor. -internal sealed record WorkflowExecutorInfo(string ExecutorId, bool IsAgenticExecutor); +public sealed record WorkflowExecutorInfo(string ExecutorId, bool IsAgenticExecutor); /// /// Represents a level of executors that can be executed in parallel (Fan-Out). @@ -19,12 +19,12 @@ internal sealed record WorkflowExecutorInfo(string ExecutorId, bool IsAgenticExe /// The level number (0-based, starting from the root executor). /// The executors that can run in parallel at this level. /// Indicates if this level is a Fan-In point (has executors with multiple predecessors). -internal sealed record WorkflowExecutionLevel(int Level, List Executors, bool IsFanIn); +public sealed record WorkflowExecutionLevel(int Level, List Executors, bool IsFanIn); /// /// Represents the complete execution plan for a workflow, including parallel execution levels. /// -internal sealed class WorkflowExecutionPlan +public sealed class WorkflowExecutionPlan { /// /// The execution levels in order. Each level contains executors that can run in parallel. @@ -52,7 +52,10 @@ internal sealed class WorkflowExecutionPlan public bool HasFanIn => this.Levels.Any(l => l.IsFanIn); } -internal static class WorkflowHelper +/// +/// Provides helper methods for analyzing and executing workflows. +/// +public static class WorkflowHelper { /// /// Accepts a workflow instance and returns a list of executors with metadata in the order they should be executed. @@ -199,7 +202,7 @@ internal static class WorkflowHelper /// /// The executor type to check. /// true if the executor is an agentic executor; otherwise, false. - private static bool IsAgentExecutorType(TypeId executorType) + internal static bool IsAgentExecutorType(TypeId executorType) { // hack for now. In the future, the MAF type could expose something which can help with this. // Check if the type name or assembly indicates it's an agent executor diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/WorkflowSharedStateEntity.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowSharedStateEntity.cs similarity index 94% rename from dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/WorkflowSharedStateEntity.cs rename to dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowSharedStateEntity.cs index f23a4e0840..1314edcb31 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/WorkflowSharedStateEntity.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowSharedStateEntity.cs @@ -2,7 +2,7 @@ using Microsoft.DurableTask.Entities; -namespace Microsoft.Agents.AI.Hosting.AzureFunctions; +namespace Microsoft.Agents.AI.DurableTask; /// /// Durable entity that manages workflow state across activities within an orchestration run. @@ -10,7 +10,7 @@ namespace Microsoft.Agents.AI.Hosting.AzureFunctions; /// ensuring state isolation between workflow runs. The entity is automatically cleaned up /// when the orchestration completes. /// -internal sealed class WorkflowSharedStateEntity : TaskEntity +public sealed class WorkflowSharedStateEntity : TaskEntity { /// /// The entity name used for registration and lookup. @@ -125,7 +125,7 @@ internal sealed class WorkflowSharedStateEntity : TaskEntity /// /// Represents the internal state data for a workflow state entity. /// -internal sealed class WorkflowStateData +public sealed class WorkflowStateData { /// /// Gets the state dictionary mapping scope-prefixed keys to serialized values. @@ -136,7 +136,7 @@ internal sealed class WorkflowStateData /// /// Request model for reading workflow state. /// -internal sealed class WorkflowStateReadRequest +public sealed class WorkflowStateReadRequest { /// /// Gets or sets the state key. @@ -152,7 +152,7 @@ internal sealed class WorkflowStateReadRequest /// /// Request model for writing workflow state. /// -internal sealed class WorkflowStateWriteRequest +public sealed class WorkflowStateWriteRequest { /// /// Gets or sets the state key. diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/BuiltInFunctions.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/BuiltInFunctions.cs index 08f484a7a8..1851f5eff4 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/BuiltInFunctions.cs +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/BuiltInFunctions.cs @@ -42,7 +42,7 @@ internal static class BuiltInFunctions string activityFunctionName = functionContext.FunctionDefinition.Name; - DurableWorkflowRunner runner = functionContext.InstanceServices.GetRequiredService(); + FunctionsWorkflowRunner runner = functionContext.InstanceServices.GetRequiredService(); return runner.ExecuteActivityAsync(activityFunctionName, input, durableTaskClient, functionContext); } @@ -78,7 +78,7 @@ internal static class BuiltInFunctions var workflowName = context.FunctionDefinition.Name.Replace("http-", ""); var orchestrationFunctionName = $"dafx-{workflowName}"; var inputMessage = await req.ReadAsStringAsync(); - string instanceId = await client.ScheduleNewOrchestrationInstanceAsync(orchestrationFunctionName, new DuableWorkflowRunRequest { WorkflowName = workflowName, Input = inputMessage! }); //OrchFunction"); // dafx-MyTestWorkflow"); + string instanceId = await client.ScheduleNewOrchestrationInstanceAsync(orchestrationFunctionName, new DurableWorkflowRunRequest { WorkflowName = workflowName, Input = inputMessage! }); //OrchFunction"); // dafx-MyTestWorkflow"); HttpResponseData response = req.CreateResponse(HttpStatusCode.Accepted); await response.WriteStringAsync($"InvokeWorkflowOrechstrtationAsync is invoked for {workflowName}. Orchestration instanceId: {instanceId}"); @@ -221,7 +221,7 @@ internal static class BuiltInFunctions } #pragma warning disable DURTASK001 // Durable analyzer complained - public static Task> WorkflowRunnerOrchestrationAsync(TaskOrchestrationContext context, DuableWorkflowRunRequest input) + public static Task> WorkflowRunnerOrchestrationAsync(TaskOrchestrationContext context, DurableWorkflowRunRequest input) { ArgumentNullException.ThrowIfNull(context); diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/CoreAgentConfigurationExtensions.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/CoreAgentConfigurationExtensions.cs index ff5c77416c..16ae42662c 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/CoreAgentConfigurationExtensions.cs +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/CoreAgentConfigurationExtensions.cs @@ -39,7 +39,7 @@ internal static class CoreAgentConfigurationExtensions /// The functions application builder for method chaining. internal static FunctionsApplicationBuilder RegisterWorkflowServices(this FunctionsApplicationBuilder builder) { - builder.Services.TryAddSingleton(); + builder.Services.TryAddSingleton(); builder.Services.TryAddEnumerable(ServiceDescriptor.Singleton()); return builder; diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableOptionsExtensions.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableOptionsExtensions.cs index f946f4695e..6ec90142e6 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableOptionsExtensions.cs +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableOptionsExtensions.cs @@ -91,7 +91,7 @@ public static class DurableOptionsExtensions foreach (string workflowName in workflows.Workflows.Select(kp => kp.Key)) { - tasks.AddOrchestratorFunc>( + tasks.AddOrchestratorFunc>( $"dafx-{workflowName}", async (orchestrationContext, request) => { diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowFunctionMetadataTransformer.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowFunctionMetadataTransformer.cs index a591da793d..abdcd2b020 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowFunctionMetadataTransformer.cs +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowFunctionMetadataTransformer.cs @@ -64,7 +64,7 @@ internal sealed class DurableWorkflowFunctionMetadataTransformer : IFunctionMeta string functionName = $"dafx-{executorId.Split("_")[0]}"; // Check if the executor type is an agent-related type - if (IsAgentExecutorType(executorInfo.ExecutorType)) + if (WorkflowHelper.IsAgentExecutorType(executorInfo.ExecutorType)) { this._logger.LogAddingAgentEntityFunction(executorId, executorInfo.ExecutorType.TypeName, workflow.Key); //original.Add(CreateAgentTrigger(functionName)); @@ -79,18 +79,6 @@ internal sealed class DurableWorkflowFunctionMetadataTransformer : IFunctionMeta } this._logger.LogTransformFinished(original.Count); - - static bool IsAgentExecutorType(TypeId executorType) - { - // hack for now. In the future, the MAF type could expose something which can help with this. - // Check if the type name or assembly indicates it's an agent executor - // This includes AgentRunStreamingExecutor, AgentExecutor, ChatClientAgent wrappers, etc. - string typeName = executorType.TypeName; - string assemblyName = executorType.AssemblyName; - - return typeName.Contains("AIAgentHostExecutor", StringComparison.OrdinalIgnoreCase) && - assemblyName.Contains("Microsoft.Agents.AI", StringComparison.OrdinalIgnoreCase); - } } private static DefaultFunctionMetadata CreateHttpTrigger(string name, string route) diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowRunner.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowRunner.cs index 7713676a15..9b4df91b9e 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowRunner.cs +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowRunner.cs @@ -1,79 +1,29 @@ // Copyright (c) Microsoft. All rights reserved. using System.Diagnostics.CodeAnalysis; -using System.Text.Json; using Microsoft.Agents.AI.DurableTask; using Microsoft.Agents.AI.Workflows; using Microsoft.Agents.AI.Workflows.Checkpointing; using Microsoft.Azure.Functions.Worker; -using Microsoft.DurableTask; using Microsoft.DurableTask.Client; -using Microsoft.DurableTask.Entities; using Microsoft.Extensions.Logging; namespace Microsoft.Agents.AI.Hosting.AzureFunctions; /// -/// Executes workflow orchestrations and activity functions for durable workflows. +/// Azure Functions-specific workflow runner that extends the base +/// with Azure Functions activity execution support. /// -internal sealed class DurableWorkflowRunner +internal sealed class FunctionsWorkflowRunner : DurableWorkflowRunner { - private readonly DurableWorkflowOptions _options; - private readonly ILogger _logger; - /// - /// Initializes a new instance of the class. + /// Initializes a new instance of the class. /// /// The logger instance. /// The durable options containing workflow configurations. - public DurableWorkflowRunner(ILogger logger, DurableOptions durableOptions) + public FunctionsWorkflowRunner(ILogger logger, DurableOptions durableOptions) + : base(logger, durableOptions) { - ArgumentNullException.ThrowIfNull(logger); - ArgumentNullException.ThrowIfNull(durableOptions); - - this._logger = logger; - this._options = durableOptions.Workflows; - } - - /// - /// Runs a workflow orchestration. - /// - /// The task orchestration context. - /// The workflow run request containing workflow name and input. - /// The replay-safe logger for orchestration logging. - /// A list containing the workflow execution result. - internal async Task> RunWorkflowOrchestrationAsync( - TaskOrchestrationContext context, - DuableWorkflowRunRequest request, - ILogger logger) - { - ArgumentNullException.ThrowIfNull(context); - ArgumentNullException.ThrowIfNull(request); - - if (!this._options.Workflows.TryGetValue(request.WorkflowName, out Workflow? workflow)) - { - throw new InvalidOperationException($"Workflow '{request.WorkflowName}' not found."); - } - - logger.LogRunningWorkflow(workflow.Name); - - string result = await this.ExecuteWorkflowLevelsAsync(context, workflow, request.Input, logger).ConfigureAwait(true); - - await CleanupWorkflowStateAsync(context).ConfigureAwait(true); - - return [result]; - } - - /// - /// Cleans up the workflow state entity by signaling it to delete itself. - /// - private static async Task CleanupWorkflowStateAsync(TaskOrchestrationContext context) - { - EntityInstanceId stateEntityId = new(WorkflowSharedStateEntity.EntityName, context.InstanceId); - - // Call the entity's Delete method to clean up state - // Using CallEntityAsync ensures the deletion completes before the orchestration finishes - await context.Entities.CallEntityAsync(stateEntityId, nameof(WorkflowSharedStateEntity.Delete)).ConfigureAwait(true); } /// @@ -97,12 +47,12 @@ internal sealed class DurableWorkflowRunner string executorName = ParseExecutorName(activityFunctionName); - if (!this._options.Executors.TryGetExecutor(executorName, out ExecutorRegistration? registration) || registration is null) + if (!this.Options.Executors.TryGetExecutor(executorName, out ExecutorRegistration? registration) || registration is null) { throw new InvalidOperationException($"Executor '{executorName}' not found in the executor registry."); } - this._logger.LogExecutingActivity(registration.ExecutorId, executorName); + this.Logger.LogExecutingActivity(registration.ExecutorId, executorName); Executor executor = await registration.CreateExecutorInstanceAsync("activity-run", CancellationToken.None) .ConfigureAwait(false); @@ -147,223 +97,4 @@ internal sealed class DurableWorkflowRunner { return new DurableExecutorContext(instanceId, client); } - - private async Task ExecuteWorkflowLevelsAsync( - TaskOrchestrationContext context, - Workflow workflow, - string initialInput, - ILogger logger) - { - WorkflowExecutionPlan plan = WorkflowHelper.GetExecutionPlan(workflow); - Dictionary results = []; - - foreach (WorkflowExecutionLevel level in plan.Levels) - { - if (level.Executors.Count == 1) - { - WorkflowExecutorInfo executorInfo = level.Executors[0]; - string input = GetExecutorInput(executorInfo.ExecutorId, initialInput, results, plan); - results[executorInfo.ExecutorId] = await this.ExecuteExecutorAsync(context, executorInfo, input, logger).ConfigureAwait(true); - } - else - { - List> tasks = []; - foreach (WorkflowExecutorInfo executorInfo in level.Executors) - { - string input = GetExecutorInput(executorInfo.ExecutorId, initialInput, results, plan); - tasks.Add(this.ExecuteExecutorWithIdAsync(context, executorInfo, input, logger)); - } - - foreach ((string id, string result) in await Task.WhenAll(tasks).ConfigureAwait(true)) - { - results[id] = result; - } - } - } - - return GetFinalResult(plan, results); - } - - private async Task<(string Id, string Result)> ExecuteExecutorWithIdAsync( - TaskOrchestrationContext context, - WorkflowExecutorInfo executorInfo, - string input, - ILogger logger) - { - string result = await this.ExecuteExecutorAsync(context, executorInfo, input, logger).ConfigureAwait(true); - return (executorInfo.ExecutorId, result); - } - - private async Task ExecuteExecutorAsync( - TaskOrchestrationContext context, - WorkflowExecutorInfo executorInfo, - string input, - ILogger logger) - { - if (!executorInfo.IsAgenticExecutor) - { - string triggerName = $"dafx-{GetBaseName(executorInfo.ExecutorId)}"; - return await context.CallActivityAsync(triggerName, input).ConfigureAwait(true); - } - - return await ExecuteAgentAsync(context, executorInfo, input, logger).ConfigureAwait(true); - } - - private static async Task ExecuteAgentAsync( - TaskOrchestrationContext context, - WorkflowExecutorInfo executorInfo, - string input, - ILogger logger) - { - string agentName = GetBaseName(executorInfo.ExecutorId); - DurableAIAgent agent = context.GetAgent(agentName); - - if (agent is null) - { - logger.LogWarning("Agent '{AgentName}' not found", agentName); - return $"Agent '{agentName}' not found"; - } - - AgentThread thread = agent.GetNewThread(); - AgentRunResponse response = await agent.RunAsync(input, thread).ConfigureAwait(true); - return response.Text; - } - - private static string GetExecutorInput( - string executorId, - string initialInput, - Dictionary results, - WorkflowExecutionPlan plan) - { - List predecessors = plan.Predecessors[executorId]; - - if (predecessors.Count == 0) - { - return initialInput; - } - - if (predecessors.Count == 1) - { - return results.TryGetValue(predecessors[0], out string? result) ? result : initialInput; - } - - List aggregated = []; - foreach (string predecessorId in predecessors) - { - if (results.TryGetValue(predecessorId, out string? result)) - { - aggregated.Add(result); - } - } - - return SerializeToJson(aggregated); - } - - private static string GetFinalResult(WorkflowExecutionPlan plan, Dictionary results) - { - WorkflowExecutionLevel lastLevel = plan.Levels[^1]; - - if (lastLevel.Executors.Count == 1) - { - return results[lastLevel.Executors[0].ExecutorId]; - } - - List finalResults = []; - foreach (WorkflowExecutorInfo executor in lastLevel.Executors) - { - if (results.TryGetValue(executor.ExecutorId, out string? result)) - { - finalResults.Add(result); - } - } - - return string.Join("\n---\n", finalResults); - } - - private static string ParseExecutorName(string activityFunctionName) - { - const string Prefix = "dafx-"; - - if (!activityFunctionName.StartsWith(Prefix, StringComparison.Ordinal)) - { - throw new InvalidOperationException( - $"Activity function name '{activityFunctionName}' does not start with '{Prefix}' prefix."); - } - - string executorName = activityFunctionName[Prefix.Length..]; - - if (string.IsNullOrEmpty(executorName)) - { - throw new InvalidOperationException( - $"Activity function name '{activityFunctionName}' is not in the expected format '{Prefix}{{executorName}}'."); - } - - return executorName; - } - - private static string GetBaseName(string executorId) - { - int underscoreIndex = executorId.IndexOf('_'); - return underscoreIndex > 0 ? executorId[..underscoreIndex] : executorId; - } - - [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Serializing known types.")] - [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Serializing known types.")] - private static string SerializeToJson(List values) - { - return JsonSerializer.Serialize(values); - } - - [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Serializing workflow types registered at startup.")] - [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Serializing workflow types registered at startup.")] - private 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); - } - - [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow types registered at startup.")] - [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing workflow types registered at startup.")] - private static object DeserializeInput(string input, Type targetType) - { - if (targetType == typeof(string)) - { - return input; - } - - string json = input; - if (input.StartsWith('"') && input.EndsWith('"')) - { - try - { - string? innerJson = JsonSerializer.Deserialize(input); - if (innerJson is not null) - { - json = innerJson; - } - } - catch (JsonException) - { - // Not double-serialized, use original - } - } - - return JsonSerializer.Deserialize(json, targetType) - ?? throw new InvalidOperationException($"Failed to deserialize input to type '{targetType.Name}'."); - } } diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowRunnerLogs.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowRunnerLogs.cs deleted file mode 100644 index 037d76f170..0000000000 --- a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowRunnerLogs.cs +++ /dev/null @@ -1,38 +0,0 @@ -// Copyright (c) Microsoft. All rights reserved. - -using System.Diagnostics.CodeAnalysis; -using Microsoft.Extensions.Logging; - -namespace Microsoft.Agents.AI.Hosting.AzureFunctions; - -/// -/// Logging messages for . -/// -[ExcludeFromCodeCoverage] -internal static partial class DurableWorkflowRunnerLogs -{ - [LoggerMessage( - Level = LogLevel.Information, - Message = "Attempting to run workflow: {WorkflowName}")] - public static partial void LogAttemptingToRunWorkflow(this ILogger logger, string workflowName); - - [LoggerMessage( - Level = LogLevel.Information, - Message = "Running workflow: {WorkflowName}")] - public static partial void LogRunningWorkflow(this ILogger logger, string? workflowName); - - [LoggerMessage( - Level = LogLevel.Information, - Message = "Attempting to execute activity in workflow '{WorkflowName}' for executor '{ExecutorName}'")] - public static partial void LogAttemptingToExecuteActivity(this ILogger logger, string workflowName, string executorName); - - [LoggerMessage( - Level = LogLevel.Information, - Message = "Executing activity for executor '{ExecutorId}' of type '{ExecutorType}'")] - public static partial void LogExecutingActivity(this ILogger logger, string executorId, string executorType); - - [LoggerMessage( - Level = LogLevel.Information, - Message = "Activity executed for executor '{ExecutorId}' with result: {Result}")] - public static partial void LogActivityExecuted(this ILogger logger, string executorId, string result); -} diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/MinimalActivityContext.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/MinimalActivityContext.cs deleted file mode 100644 index e69de29bb2..0000000000