diff --git a/dotnet/samples/AzureFunctions/12_WorkflowSharedState/12_WorkflowSharedState.csproj b/dotnet/samples/AzureFunctions/12_WorkflowSharedState/12_WorkflowSharedState.csproj new file mode 100644 index 0000000000..ec1dca7683 --- /dev/null +++ b/dotnet/samples/AzureFunctions/12_WorkflowSharedState/12_WorkflowSharedState.csproj @@ -0,0 +1,44 @@ + + + net10.0 + v4 + Exe + enable + enable + + SingleAgent + SingleAgent + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/dotnet/samples/AzureFunctions/12_WorkflowSharedState/MyOrchFunction.cs b/dotnet/samples/AzureFunctions/12_WorkflowSharedState/MyOrchFunction.cs new file mode 100644 index 0000000000..903a14cabd --- /dev/null +++ b/dotnet/samples/AzureFunctions/12_WorkflowSharedState/MyOrchFunction.cs @@ -0,0 +1,64 @@ +// Copyright (c) Microsoft. All rights reserved. + +using Microsoft.Azure.Functions.Worker; +using Microsoft.Azure.Functions.Worker.Http; +using Microsoft.DurableTask; +using Microsoft.DurableTask.Client; +using Microsoft.Extensions.Logging; + +namespace SingleAgent; + +public static class MyOrchFunction +{ + [Function(nameof(MyOrchFunction))] + public static async Task> RunOrchestratorAsync( + [OrchestrationTrigger] TaskOrchestrationContext context) + { + ILogger logger = context.CreateReplaySafeLogger(nameof(MyOrchFunction)); + logger.LogInformation("Saying hello."); + var outputs = new List(); + + // Replace name and input with values relevant for your Durable Functions Activity + outputs.Add(await context.CallActivityAsync(nameof(SayHello), "Tokyo")); + outputs.Add(await context.CallActivityAsync(nameof(SayHello), "Seattle")); + outputs.Add(await context.CallActivityAsync(nameof(SayHello), "London")); + + // returns ["Hello Tokyo!", "Hello Seattle!", "Hello London!"] + return outputs; + } + + [Function(nameof(SayHello))] + public static string SayHello([ActivityTrigger] string name, + TaskActivityContext taskActivityContext, + FunctionContext functionContext) + { + ILogger logger = functionContext.GetLogger("SayHello"); + if (logger.IsEnabled(LogLevel.Information)) + { + var instanceId = (string)functionContext.BindingContext.BindingData["instanceId"]!; + logger.LogInformation("Saying hello to {Name} for instanceId:{InstanceId}", name, instanceId); + } + return $"Hello {name}!"; + } + + [Function("MyOrchFunction_HttpStart")] + public static async Task HttpStartAsync( + [HttpTrigger(AuthorizationLevel.Anonymous, "get", "post")] HttpRequestData req, + [DurableClient] DurableTaskClient client, + FunctionContext executionContext) + { + ILogger logger = executionContext.GetLogger("MyOrchFunction_HttpStart"); + + // Function input comes from the request content. + string instanceId = await client.ScheduleNewOrchestrationInstanceAsync( + nameof(MyOrchFunction)); + if (logger.IsEnabled(LogLevel.Information)) + { + logger.LogInformation("Started orchestration with ID = '{InstanceId}'.", instanceId); + } + + // Returns an HTTP 202 response with an instance management payload. + // See https://learn.microsoft.com/azure/azure-functions/durable/durable-functions-http-api#start-orchestration + return await client.CreateCheckStatusResponseAsync(req, instanceId); + } +} diff --git a/dotnet/samples/AzureFunctions/12_WorkflowSharedState/OrderIdParserExecutor.cs b/dotnet/samples/AzureFunctions/12_WorkflowSharedState/OrderIdParserExecutor.cs new file mode 100644 index 0000000000..d79455ff0d --- /dev/null +++ b/dotnet/samples/AzureFunctions/12_WorkflowSharedState/OrderIdParserExecutor.cs @@ -0,0 +1,87 @@ +// Copyright (c) Microsoft. All rights reserved. + +// This sample demonstrates how to use durable state management in Azure Functions workflows. +// The OrderIdParserExecutor writes a value to shared state, and the EmailSenderExecutor reads it back. +// The state is persisted durably using Durable Entities behind the scenes. + +using Microsoft.Agents.AI.Workflows; + +namespace SingleAgent; + +/// +/// Constants for shared state scopes used across executors. +/// +internal static class SharedStateConstants +{ + public const string MessageScope = "MessageState"; + public const string ProcessedMessageKey = "ProcessedMessage"; +} + +public sealed class Order +{ + public Order(string id, decimal amount) + { + this.Id = id; + this.Amount = amount; + } + public string Id { get; } + public decimal Amount { get; } + public string? PaymentReferenceNumber { get; set; } +} + +/// +/// First executor that processes a message and stores the result in shared state. +/// +internal sealed class OrderIdParserExecutor() : Executor("OrderIdParserExecutor") +{ + public override async ValueTask HandleAsync(string message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + // Process the message + string processedMessage = $"Processed: {message}"; + + // Store the processed message in shared state for the next executor + await context.QueueStateUpdateAsync( + SharedStateConstants.ProcessedMessageKey, + processedMessage, + SharedStateConstants.MessageScope, + cancellationToken); + + return GetOrder(message); + } + + private static Order GetOrder(string id) + { + // Simulate fetching order details + return new Order(id, 100.0m); + } +} + +/// +/// Second executor that reads the shared state and appends to the message. +/// +internal sealed class EmailSenderExecutor() : Executor("EmailSenderExecutor") +{ + public override async ValueTask HandleAsync(Order message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + // Read the processed message from shared state (written by OrderIdParserExecutor) + string? storedMessage = await context.ReadStateAsync( + SharedStateConstants.ProcessedMessageKey, + SharedStateConstants.MessageScope, + cancellationToken); + + // Combine with the input message + return storedMessage is not null + ? $"From state: [{storedMessage}] | Input: [{message.Id}]" + : $"No state found | Input: [{message.Id}]"; + } +} + +internal sealed class PaymentProcesserExecutor() : Executor("PaymentProcesserExecutor") +{ + public override async ValueTask HandleAsync(Order message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + // Call payment gateway. + message.PaymentReferenceNumber = Guid.NewGuid().ToString().Substring(0, 4); + return message; + } +} diff --git a/dotnet/samples/AzureFunctions/12_WorkflowSharedState/Program.cs b/dotnet/samples/AzureFunctions/12_WorkflowSharedState/Program.cs new file mode 100644 index 0000000000..167e1fe6cf --- /dev/null +++ b/dotnet/samples/AzureFunctions/12_WorkflowSharedState/Program.cs @@ -0,0 +1,29 @@ +// Copyright (c) Microsoft. All rights reserved. + +using Microsoft.Agents.AI.Hosting.AzureFunctions; +using Microsoft.Agents.AI.Workflows; +using Microsoft.Azure.Functions.Worker.Builder; +using Microsoft.Extensions.Hosting; +using SingleAgent; + +// Set up an AI agent following the standard Microsoft Agent Framework pattern. + +OrderIdParserExecutor orderParser = new(); +PaymentProcesserExecutor paymentProcessor = new(); +EmailSenderExecutor emailSender = new(); + +WorkflowBuilder builder = new(orderParser); +builder.AddEdge(orderParser, paymentProcessor); +builder.AddEdge(paymentProcessor, emailSender).WithOutputFrom(emailSender); +var workflow = builder.WithName("ProcessOrder").Build(); + +FunctionsApplication.CreateBuilder(args) + .ConfigureFunctionsWebApplication() + .ConfigureDurableOptions(options => + { + options.Workflows.AddWorkflow(workflow); + + // Optional - Configure AI agents + // options.Agents.AddAIAgent(agent); + }) + .Build().Run(); diff --git a/dotnet/samples/AzureFunctions/12_WorkflowSharedState/README.md b/dotnet/samples/AzureFunctions/12_WorkflowSharedState/README.md new file mode 100644 index 0000000000..d4ac968978 --- /dev/null +++ b/dotnet/samples/AzureFunctions/12_WorkflowSharedState/README.md @@ -0,0 +1,89 @@ +# Single Agent Sample + +This sample demonstrates how to use the Durable Agent Framework (DAFx) to create a simple Azure Functions app that hosts a single AI agent and provides direct HTTP API access for interactive conversations. + +## Key Concepts Demonstrated + +- Using the Microsoft Agent Framework to define a simple AI agent with a name and instructions. +- Registering agents with the Function app and running them using HTTP. +- Conversation management (via session IDs) for isolated interactions. + +## Environment Setup + +See the [README.md](../README.md) file in the parent directory for more information on how to configure the environment, including how to install and run common sample dependencies. + +## Running the Sample + +With the environment setup and function app running, you can test the sample by sending an HTTP request to the agent endpoint. + +You can use the `demo.http` file to send a message to the agent, or a command line tool like `curl` as shown below: + +Bash (Linux/macOS/WSL): + +```bash +curl -X POST http://localhost:7071/api/agents/Joker/run \ + -H "Content-Type: text/plain" \ + -d "Tell me a joke about a pirate." +``` + +PowerShell: + +```powershell +Invoke-RestMethod -Method Post ` + -Uri http://localhost:7071/api/agents/Joker/run ` + -ContentType text/plain ` + -Body "Tell me a joke about a pirate." +``` + +You can also send JSON requests: + +```bash +curl -X POST http://localhost:7071/api/agents/Joker/run \ + -H "Content-Type: application/json" \ + -H "Accept: application/json" \ + -d '{"message": "Tell me a joke about a pirate."}' +``` + +To continue a conversation, include the `thread_id` in the query string or JSON body: + +```bash +curl -X POST "http://localhost:7071/api/agents/Joker/run?thread_id=your-thread-id" \ + -H "Content-Type: application/json" \ + -H "Accept: application/json" \ + -d '{"message": "Tell me another one."}' +``` + +The response from the agent will be displayed in the terminal where you ran `func start`. The expected `text/plain` output will look something like: + +```text +Why don't pirates ever learn the alphabet? Because they always get stuck at "C"! +``` + +The expected `application/json` output will look something like: + +```json +{ + "status": 200, + "thread_id": "ee6e47a0-f24b-40b1-ade8-16fcebb9eb40", + "response": { + "Messages": [ + { + "AuthorName": "Joker", + "CreatedAt": "2025-11-11T12:00:00.0000000Z", + "Role": "assistant", + "Contents": [ + { + "Type": "text", + "Text": "Why don't pirates ever learn the alphabet? Because they always get stuck at 'C'!" + } + ] + } + ], + "Usage": { + "InputTokenCount": 78, + "OutputTokenCount": 36, + "TotalTokenCount": 114 + } + } +} +``` diff --git a/dotnet/samples/AzureFunctions/12_WorkflowSharedState/demo.http b/dotnet/samples/AzureFunctions/12_WorkflowSharedState/demo.http new file mode 100644 index 0000000000..6329902add --- /dev/null +++ b/dotnet/samples/AzureFunctions/12_WorkflowSharedState/demo.http @@ -0,0 +1,14 @@ +# Default endpoint address for local testing +@authority=http://localhost:7071 + +### Start the workflow +POST {{authority}}/api/workflows/ProcessOrder/run +Content-Type: text/plain + +123 + +### Start second workflow +POST {{authority}}/api/workflows/ProcessOrder/run +Content-Type: text/plain + +456 \ No newline at end of file diff --git a/dotnet/samples/AzureFunctions/12_WorkflowSharedState/host.json b/dotnet/samples/AzureFunctions/12_WorkflowSharedState/host.json new file mode 100644 index 0000000000..9384a0a583 --- /dev/null +++ b/dotnet/samples/AzureFunctions/12_WorkflowSharedState/host.json @@ -0,0 +1,20 @@ +{ + "version": "2.0", + "logging": { + "logLevel": { + "Microsoft.Agents.AI.DurableTask": "Information", + "Microsoft.Agents.AI.Hosting.AzureFunctions": "Information", + "DurableTask": "Information", + "Microsoft.DurableTask": "Information" + } + }, + "extensions": { + "durableTask": { + "hubName": "default", + "storageProvider": { + "type": "AzureManaged", + "connectionStringName": "DURABLE_TASK_SCHEDULER_CONNECTION_STRING" + } + } + } +} diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/BuiltInFunctionExecutor.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/BuiltInFunctionExecutor.cs index 20da9e9d12..257ee972dd 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/BuiltInFunctionExecutor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/BuiltInFunctionExecutor.cs @@ -27,10 +27,39 @@ internal sealed class BuiltInFunctionExecutor : IFunctionExecutor if (context.FunctionDefinition.EntryPoint == BuiltInFunctions.InvokeWorkflowActivityFunctionEntryPoint) { - var binding = context.FunctionDefinition.InputBindings.Values.FirstOrDefault(a => a.Name == "input"); - var input = await context.BindInputAsync(binding!); - var val = input.Value; - context.GetInvocationResult().Value = await BuiltInFunctions.InvokeWorkflowActivityAsync(val!, context); + // Bind all inputs to get the input string and DurableTaskClient + FunctionInputBindingResult? bindingResults = await functionInputBindingFeature.BindFunctionInputAsync(context); + if (bindingResults is not { Values: { } activityBindings }) + { + throw new InvalidOperationException($"Function input binding failed for the invocation {context.InvocationId}"); + } + + DurableTaskClient? activityDurableTaskClient = null; + string? activityInput = null; + foreach (object? binding in activityBindings) + { + if (binding is string stringInput) + { + activityInput = stringInput; + } + + if (binding is DurableTaskClient client) + { + activityDurableTaskClient = client; + } + } + + if (activityInput is null) + { + throw new InvalidOperationException($"Activity input binding is missing for the invocation {context.InvocationId}."); + } + + if (activityDurableTaskClient is null) + { + throw new InvalidOperationException($"DurableTaskClient binding is missing for the invocation {context.InvocationId}."); + } + + context.GetInvocationResult().Value = await BuiltInFunctions.InvokeWorkflowActivityAsync(activityInput, activityDurableTaskClient, context); return; } diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/BuiltInFunctions.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/BuiltInFunctions.cs index 879c94ba8a..ddcfd71ce3 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/BuiltInFunctions.cs +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/BuiltInFunctions.cs @@ -33,15 +33,17 @@ internal static class BuiltInFunctions // Exposed as an activity trigger for workflow executors public static Task InvokeWorkflowActivityAsync( [ActivityTrigger] string input, + [DurableClient] DurableTaskClient durableTaskClient, FunctionContext functionContext) { ArgumentNullException.ThrowIfNull(input); + ArgumentNullException.ThrowIfNull(durableTaskClient); ArgumentNullException.ThrowIfNull(functionContext); string activityFunctionName = functionContext.FunctionDefinition.Name; DurableWorkflowRunner runner = functionContext.InstanceServices.GetRequiredService(); - return runner.ExecuteActivityAsync(activityFunctionName, input, functionContext); + return runner.ExecuteActivityAsync(activityFunctionName, input, durableTaskClient, functionContext); } // Exposed as an entity trigger via AgentFunctionsProvider @@ -74,8 +76,9 @@ internal static class BuiltInFunctions FunctionContext context) { var workflowName = context.FunctionDefinition.Name.Replace("http-", ""); + var orchestrationFunctionName = $"dafx-{workflowName}"; var inputMessage = await req.ReadAsStringAsync(); - string instanceId = await client.ScheduleNewOrchestrationInstanceAsync("WorkflowRunnerOrchestration", new DuableWorkflowRunRequest { WorkflowName = workflowName, Input = inputMessage! }); //OrchFunction"); // dafx-MyTestWorkflow"); + string instanceId = await client.ScheduleNewOrchestrationInstanceAsync(orchestrationFunctionName, new DuableWorkflowRunRequest { WorkflowName = workflowName, Input = inputMessage! }); //OrchFunction"); // dafx-MyTestWorkflow"); HttpResponseData response = req.CreateResponse(HttpStatusCode.Accepted); await response.WriteStringAsync($"InvokeWorkflowOrechstrtationAsync is invoked for {workflowName}. Orchestration instanceId: {instanceId}"); diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableExecutorContext.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableExecutorContext.cs new file mode 100644 index 0000000000..e247ed2cd0 --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableExecutorContext.cs @@ -0,0 +1,269 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Diagnostics.CodeAnalysis; +using System.Text.Json; +using Microsoft.Agents.AI.Workflows; +using Microsoft.DurableTask.Client; +using Microsoft.DurableTask.Client.Entities; +using Microsoft.DurableTask.Entities; + +namespace Microsoft.Agents.AI.Hosting.AzureFunctions; + +/// +/// An implementation of for workflow executors running as Azure Functions 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. +/// +/// +/// State operations use GetEntityAsync for reads (fetches current entity state) and SignalEntityAsync +/// for writes. Since activities run sequentially in the orchestration and entity signals are processed +/// in order, state consistency is maintained across executors. +/// +[RequiresUnreferencedCode("State serialization uses reflection-based JSON serialization.")] +[RequiresDynamicCode("State serialization uses reflection-based JSON serialization.")] +internal sealed class DurableExecutorContext : IWorkflowContext +{ + private readonly string _instanceId; + private readonly DurableTaskClient _client; + private readonly Dictionary _pendingUpdates = []; + private readonly HashSet _clearedScopes = []; + + /// + /// Initializes a new instance of the class. + /// + /// The orchestration instance ID used to scope the state entity. + /// The durable task client for entity operations. + public DurableExecutorContext(string instanceId, DurableTaskClient client) + { + this._instanceId = instanceId; + this._client = client; + } + + /// + public ValueTask AddEventAsync(WorkflowEvent workflowEvent, CancellationToken cancellationToken = default) + { + // In activity context, events are not propagated to the workflow + return default; + } + + /// + public ValueTask SendMessageAsync(object message, string? targetId = null, CancellationToken cancellationToken = default) + { + // In activity context, messages cannot be routed to other executors + return default; + } + + /// + public ValueTask YieldOutputAsync(object output, CancellationToken cancellationToken = default) + { + // In activity context, outputs are not yielded to the workflow + return default; + } + + /// + public ValueTask RequestHaltAsync() + { + // Halt requests are not supported in activity context + return default; + } + + /// + public async ValueTask ReadStateAsync(string key, string? scopeName = null, CancellationToken cancellationToken = default) + { + string scopeKey = GetScopeKey(scopeName, key); + + // 1. Check pending updates first (read-your-writes within this activity) + if (this._pendingUpdates.TryGetValue(scopeKey, out string? pendingValue)) + { + return pendingValue is null ? default : JsonSerializer.Deserialize(pendingValue); + } + + // 2. Check if the scope was cleared in this activity + string normalizedScope = scopeName ?? "__default__"; + if (this._clearedScopes.Contains(normalizedScope)) + { + return default; + } + + // 3. Read from the durable entity + EntityInstanceId entityId = this.GetStateEntityId(); + + EntityMetadata? metadata = await this._client.Entities + .GetEntityAsync(entityId, includeState: true, cancellation: cancellationToken) + .ConfigureAwait(false); + + if (metadata?.IncludesState != true) + { + return default; + } + + WorkflowStateData? stateData = metadata.State.ReadAs(); + if (stateData?.Values is null) + { + return default; + } + + if (stateData.Values.TryGetValue(scopeKey, out string? serializedValue) && serializedValue is not null) + { + return JsonSerializer.Deserialize(serializedValue); + } + + return default; + } + + /// + public async ValueTask ReadOrInitStateAsync(string key, Func initialStateFactory, string? scopeName = null, CancellationToken cancellationToken = default) + { + T? value = await this.ReadStateAsync(key, scopeName, cancellationToken).ConfigureAwait(false); + + if (value is not null) + { + return value; + } + + // Initialize with factory value and write to entity + T initialValue = initialStateFactory(); + await this.QueueStateUpdateAsync(key, initialValue, scopeName, cancellationToken).ConfigureAwait(false); + return initialValue; + } + + /// + public async ValueTask> ReadStateKeysAsync(string? scopeName = null, CancellationToken cancellationToken = default) + { + string normalizedScope = scopeName ?? "__default__"; + string scopePrefix = GetScopePrefix(scopeName); + HashSet keys = []; + + // If scope was cleared, only return keys from pending updates + if (this._clearedScopes.Contains(normalizedScope)) + { + return this.GetPendingKeysForScope(scopeName); + } + + // Read keys from the durable entity + EntityInstanceId entityId = this.GetStateEntityId(); + + EntityMetadata? metadata = await this._client.Entities + .GetEntityAsync(entityId, includeState: true, cancellation: cancellationToken) + .ConfigureAwait(false); + + if (metadata?.IncludesState == true) + { + WorkflowStateData? stateData = metadata.State.ReadAs(); + if (stateData?.Values is not null) + { + foreach (string scopeKey in stateData.Values.Keys) + { + if (scopeKey.StartsWith(scopePrefix, StringComparison.Ordinal)) + { + string foundKey = scopeKey[scopePrefix.Length..]; + keys.Add(foundKey); + } + } + } + } + + // Merge with pending updates + foreach (KeyValuePair pending in this._pendingUpdates) + { + if (pending.Key.StartsWith(scopePrefix, StringComparison.Ordinal)) + { + string foundKey = pending.Key[scopePrefix.Length..]; + if (pending.Value is not null) + { + keys.Add(foundKey); + } + else + { + keys.Remove(foundKey); + } + } + } + + return keys; + } + + /// + public async ValueTask QueueStateUpdateAsync(string key, T? value, string? scopeName = null, CancellationToken cancellationToken = default) + { + string scopeKey = GetScopeKey(scopeName, key); + string? serializedValue = value is null ? null : JsonSerializer.Serialize(value); + + // Store locally for read-your-writes within this activity + this._pendingUpdates[scopeKey] = serializedValue; + + // Write to the durable entity via signal + // Since activities run sequentially and signals are processed in order, + // the next activity will see this update when it reads from the entity + EntityInstanceId entityId = this.GetStateEntityId(); + WorkflowStateWriteRequest request = new() { Key = key, ScopeName = scopeName, Value = serializedValue }; + + await this._client.Entities + .SignalEntityAsync(entityId, nameof(WorkflowSharedStateEntity.WriteState), request, cancellation: cancellationToken) + .ConfigureAwait(false); + } + + /// + public async ValueTask QueueClearScopeAsync(string? scopeName = null, CancellationToken cancellationToken = default) + { + string normalizedScope = scopeName ?? "__default__"; + this._clearedScopes.Add(normalizedScope); + + // Remove pending updates in this scope + string scopePrefix = GetScopePrefix(scopeName); + List keysToRemove = this._pendingUpdates.Keys + .Where(k => k.StartsWith(scopePrefix, StringComparison.Ordinal)) + .ToList(); + + foreach (string key in keysToRemove) + { + this._pendingUpdates.Remove(key); + } + + // Clear in the durable entity via signal + EntityInstanceId entityId = this.GetStateEntityId(); + + await this._client.Entities + .SignalEntityAsync(entityId, nameof(WorkflowSharedStateEntity.ClearScope), scopeName, cancellation: cancellationToken) + .ConfigureAwait(false); + } + + /// + public IReadOnlyDictionary? TraceContext => null; + + /// + public bool ConcurrentRunsEnabled => false; + + private EntityInstanceId GetStateEntityId() + { + // Entity is keyed by orchestration instance ID for isolation between runs + return new EntityInstanceId(WorkflowSharedStateEntity.EntityName, this._instanceId); + } + + private HashSet GetPendingKeysForScope(string? scopeName) + { + string scopePrefix = GetScopePrefix(scopeName); + HashSet keys = []; + + foreach (KeyValuePair pending in this._pendingUpdates) + { + if (pending.Key.StartsWith(scopePrefix, StringComparison.Ordinal) && pending.Value is not null) + { + string key = pending.Key[scopePrefix.Length..]; + keys.Add(key); + } + } + + return keys; + } + + private static string GetScopeKey(string? scopeName, string key) + { + return $"{GetScopePrefix(scopeName)}{key}"; + } + + private static string GetScopePrefix(string? scopeName) + { + return scopeName is null ? "__default__:" : $"{scopeName}:"; + } +} diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableOptionsExtensions.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableOptionsExtensions.cs index 8e5c917b4f..f946f4695e 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableOptionsExtensions.cs +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableOptionsExtensions.cs @@ -81,10 +81,14 @@ public static class DurableOptionsExtensions private static void ConfigureWorkflowOrchestrations(FunctionsApplicationBuilder builder, DurableWorkflowOptions workflows) { - // Registering orchestration functions for each workflow. + // Registering orchestration functions and the workflow state entity. builder.ConfigureDurableWorker().AddTasks(tasks => { + // Register the workflow state entity for durable state management + // Each orchestration instance gets its own entity keyed by instance ID + tasks.AddEntity(WorkflowSharedStateEntity.EntityName); + foreach (string workflowName in workflows.Workflows.Select(kp => kp.Key)) { tasks.AddOrchestratorFunc>( diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowFunctionMetadataTransformer.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowFunctionMetadataTransformer.cs index 4f529eb54d..a591da793d 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowFunctionMetadataTransformer.cs +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowFunctionMetadataTransformer.cs @@ -136,6 +136,7 @@ internal sealed class DurableWorkflowFunctionMetadataTransformer : IFunctionMeta RawBindings = [ """{"name":"input","type":"activityTrigger","direction":"In","dataType":"String"}""", + """{"name":"durableTaskClient","type":"durableClient","direction":"In"}""" ], EntryPoint = BuiltInFunctions.InvokeWorkflowActivityFunctionEntryPoint, ScriptFile = BuiltInFunctions.ScriptFile, diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowRunner.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowRunner.cs index 350ead2417..7713676a15 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowRunner.cs +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowRunner.cs @@ -7,6 +7,8 @@ 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; @@ -56,23 +58,41 @@ internal sealed class DurableWorkflowRunner 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); + } + /// /// Executes an activity function for a workflow executor. /// /// The name of the activity function to execute. - /// The input string for the executor. - /// The Azure Functions context. - /// The serialized result of the executor. + /// The serialized executor input. + /// The durable task client for entity operations. + /// The function context containing binding data with the orchestration instance ID. + /// The serialized executor output. internal async Task ExecuteActivityAsync( string activityFunctionName, string input, + DurableTaskClient durableTaskClient, FunctionContext functionContext) { ArgumentNullException.ThrowIfNull(activityFunctionName); ArgumentNullException.ThrowIfNull(input); + ArgumentNullException.ThrowIfNull(durableTaskClient); ArgumentNullException.ThrowIfNull(functionContext); string executorName = ParseExecutorName(activityFunctionName); @@ -90,15 +110,44 @@ internal sealed class DurableWorkflowRunner Type inputType = executor.InputTypes.FirstOrDefault() ?? typeof(string); object typedInput = DeserializeInput(input, inputType); + // Get the orchestration instance ID from the function context binding data + string instanceId = GetInstanceIdFromContext(functionContext) + ?? throw new InvalidOperationException( + "Could not retrieve orchestration instance ID from FunctionContext. " + + "Ensure the activity is being called from within a durable orchestration."); + + // Create context with durable entity-backed state + IWorkflowContext context = CreateExecutorContext(instanceId, durableTaskClient); + object? result = await executor.ExecuteAsync( typedInput, new TypeId(inputType), - new MinimalActivityContext(registration.ExecutorId), + context, CancellationToken.None).ConfigureAwait(false); return SerializeResult(result); } + private static string? GetInstanceIdFromContext(FunctionContext functionContext) + { + if (functionContext.BindingContext.BindingData.TryGetValue("instanceId", out object? instanceIdObj) && + instanceIdObj is string instanceId) + { + return instanceId; + } + + return null; + } + + [UnconditionalSuppressMessage("AOT", "IL2026:RequiresUnreferencedCode", Justification = "DurableExecutorContext state serialization is done at runtime with user-known types.")] + [UnconditionalSuppressMessage("AOT", "IL3050:RequiresDynamicCode", Justification = "DurableExecutorContext state serialization is done at runtime with user-known types.")] + private static DurableExecutorContext CreateExecutorContext( + string instanceId, + DurableTaskClient client) + { + return new DurableExecutorContext(instanceId, client); + } + private async Task ExecuteWorkflowLevelsAsync( TaskOrchestrationContext context, Workflow workflow, diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/MinimalActivityContext.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/MinimalActivityContext.cs index 557173c498..e69de29bb2 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/MinimalActivityContext.cs +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/MinimalActivityContext.cs @@ -1,90 +0,0 @@ -// Copyright (c) Microsoft. All rights reserved. - -using Microsoft.Agents.AI.Workflows; - -namespace Microsoft.Agents.AI.Hosting.AzureFunctions; - -/// -/// A minimal implementation of for use in Azure Functions activities. -/// This provides basic context support for simple executors that don't require full workflow infrastructure. -/// -internal sealed class MinimalActivityContext : IWorkflowContext -{ - public MinimalActivityContext(string executorId) - { - // executorId is provided but not stored since this minimal context doesn't use it - _ = executorId; - } - - /// - public ValueTask AddEventAsync(WorkflowEvent workflowEvent, CancellationToken cancellationToken = default) - { - // In activity context, events are not propagated to the workflow - // They would need to be returned as part of the activity result - return default; - } - - /// - public ValueTask SendMessageAsync(object message, string? targetId = null, CancellationToken cancellationToken = default) - { - // In activity context, messages cannot be routed to other executors - // The orchestration handles message routing between executors - return default; - } - - /// - public ValueTask YieldOutputAsync(object output, CancellationToken cancellationToken = default) - { - // In activity context, outputs are not yielded to the workflow - // They would need to be returned as part of the activity result - return default; - } - - /// - public ValueTask RequestHaltAsync() - { - // Halt requests are not supported in activity context - return default; - } - - /// - public ValueTask ReadStateAsync(string key, string? scopeName = null, CancellationToken cancellationToken = default) - { - // No state available in activity context - return new ValueTask(default(T)); - } - - /// - public ValueTask ReadOrInitStateAsync(string key, Func initialStateFactory, string? scopeName = null, CancellationToken cancellationToken = default) - { - // Initialize with factory value since no state is available - return new ValueTask(initialStateFactory()); - } - - /// - public ValueTask> ReadStateKeysAsync(string? scopeName = null, CancellationToken cancellationToken = default) - { - // No state keys in activity context - return new ValueTask>([]); - } - - /// - public ValueTask QueueStateUpdateAsync(string key, T? value, string? scopeName = null, CancellationToken cancellationToken = default) - { - // State updates are not persisted in activity context - return default; - } - - /// - public ValueTask QueueClearScopeAsync(string? scopeName = null, CancellationToken cancellationToken = default) - { - // No state to clear in activity context - return default; - } - - /// - public IReadOnlyDictionary? TraceContext => null; - - /// - public bool ConcurrentRunsEnabled => false; -} diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/WorkflowSharedStateEntity.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/WorkflowSharedStateEntity.cs new file mode 100644 index 0000000000..f23a4e0840 --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/WorkflowSharedStateEntity.cs @@ -0,0 +1,171 @@ +// Copyright (c) Microsoft. All rights reserved. + +using Microsoft.DurableTask.Entities; + +namespace Microsoft.Agents.AI.Hosting.AzureFunctions; + +/// +/// Durable entity that manages workflow state across activities within an orchestration run. +/// Each orchestration instance gets its own entity instance (keyed by orchestration instance ID), +/// ensuring state isolation between workflow runs. The entity is automatically cleaned up +/// when the orchestration completes. +/// +internal sealed class WorkflowSharedStateEntity : TaskEntity +{ + /// + /// The entity name used for registration and lookup. + /// + public const string EntityName = "workflow-shared-state"; + + /// + /// Reads a state value by key and scope. + /// + /// The read request containing key and optional scope. + /// The serialized state value, or null if not found. + public string? ReadState(WorkflowStateReadRequest request) + { + string scopeKey = GetScopeKey(request.ScopeName, request.Key); + return this.State.Values.TryGetValue(scopeKey, out string? value) ? value : null; + } + + /// + /// Reads the entire state dictionary. + /// + /// A copy of the current state. + public Dictionary ReadAllState() + { + return new Dictionary(this.State.Values); + } + + /// + /// Writes or updates a state value by key and scope. + /// + /// The write request containing key, scope, and value. + public void WriteState(WorkflowStateWriteRequest request) + { + string scopeKey = GetScopeKey(request.ScopeName, request.Key); + + if (request.Value is null) + { + this.State.Values.Remove(scopeKey); + } + else + { + this.State.Values[scopeKey] = request.Value; + } + } + + /// + /// Gets all keys within a specific scope. + /// + /// The scope name, or null for the default scope. + /// A collection of keys within the scope. + public HashSet GetStateKeys(string? scopeName) + { + string scopePrefix = GetScopePrefix(scopeName); + HashSet keys = []; + + foreach (string scopeKey in this.State.Values.Keys) + { + if (scopeKey.StartsWith(scopePrefix, StringComparison.Ordinal)) + { + string key = scopeKey[scopePrefix.Length..]; + keys.Add(key); + } + } + + return keys; + } + + /// + /// Clears all state entries within a specific scope. + /// + /// The scope name, or null for the default scope. + public void ClearScope(string? scopeName) + { + string scopePrefix = GetScopePrefix(scopeName); + List keysToRemove = []; + + foreach (string scopeKey in this.State.Values.Keys) + { + if (scopeKey.StartsWith(scopePrefix, StringComparison.Ordinal)) + { + keysToRemove.Add(scopeKey); + } + } + + foreach (string key in keysToRemove) + { + this.State.Values.Remove(key); + } + } + + /// + /// Deletes the entity, cleaning up all state. + /// Called by the orchestration when it completes. + /// + public void Delete() + { + // Setting State to null tells the Durable Task framework to delete the entity. + // The entity will be garbage collected after idle timeout. + this.State = null!; + } + + private static string GetScopeKey(string? scopeName, string key) + { + return $"{GetScopePrefix(scopeName)}{key}"; + } + + private static string GetScopePrefix(string? scopeName) + { + return scopeName is null ? "__default__:" : $"{scopeName}:"; + } +} + +/// +/// Represents the internal state data for a workflow state entity. +/// +internal sealed class WorkflowStateData +{ + /// + /// Gets the state dictionary mapping scope-prefixed keys to serialized values. + /// + public Dictionary Values { get; init; } = []; +} + +/// +/// Request model for reading workflow state. +/// +internal sealed class WorkflowStateReadRequest +{ + /// + /// Gets or sets the state key. + /// + public string Key { get; set; } = string.Empty; + + /// + /// Gets or sets the optional scope name. + /// + public string? ScopeName { get; set; } +} + +/// +/// Request model for writing workflow state. +/// +internal sealed class WorkflowStateWriteRequest +{ + /// + /// Gets or sets the state key. + /// + public string Key { get; set; } = string.Empty; + + /// + /// Gets or sets the optional scope name. + /// + public string? ScopeName { get; set; } + + /// + /// Gets or sets the serialized value, or null to delete the key. + /// + public string? Value { get; set; } +}