Shared state support using durable entity.

2026-06-16 21:04:09 +08:00 · 2026-01-20 08:56:12 -08:00
parent 5ddb4cd546
commit 8c3182e4e8
15 changed files with 884 additions and 101 deletions
@@ -0,0 +1,44 @@
+<Project Sdk="Microsoft.NET.Sdk">
+  <PropertyGroup>
+    <TargetFrameworks>net10.0</TargetFrameworks>
+    <AzureFunctionsVersion>v4</AzureFunctionsVersion>
+    <OutputType>Exe</OutputType>
+    <ImplicitUsings>enable</ImplicitUsings>
+    <Nullable>enable</Nullable>
+    <!-- The Functions build tools don't like namespaces that start with a number -->
+    <AssemblyName>SingleAgent</AssemblyName>
+    <RootNamespace>SingleAgent</RootNamespace>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <FrameworkReference Include="Microsoft.AspNetCore.App" />
+  </ItemGroup>
+
+  <!-- Azure Functions packages -->
+  <ItemGroup>
+    <PackageReference Include="Microsoft.Azure.Functions.Worker" />
+    <PackageReference Include="Microsoft.Azure.Functions.Worker.Extensions.DurableTask" />
+    <PackageReference Include="Microsoft.Azure.Functions.Worker.Extensions.DurableTask.AzureManaged" />
+    <PackageReference Include="Microsoft.Azure.Functions.Worker.Extensions.Http" />
+    <PackageReference Include="Microsoft.Azure.Functions.Worker.Extensions.Http.AspNetCore" />
+    <PackageReference Include="Microsoft.Azure.Functions.Worker.Sdk" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Azure.AI.OpenAI" />
+    <PackageReference Include="Azure.Identity" />
+  </ItemGroup>
+
+  <!-- Local projects that should be switched to package references when using the sample outside of this MAF repo -->
+  <!--
+  <ItemGroup>
+    <PackageReference Include="Microsoft.Agents.AI.Hosting.AzureFunctions" />
+    <PackageReference Include="Microsoft.Agents.AI.OpenAI" />
+  </ItemGroup>
+  -->
+  <ItemGroup>
+    <ProjectReference Include="..\..\..\src\Microsoft.Agents.AI.Hosting.AzureFunctions\Microsoft.Agents.AI.Hosting.AzureFunctions.csproj" />
+    <ProjectReference Include="..\..\..\src\Microsoft.Agents.AI.OpenAI\Microsoft.Agents.AI.OpenAI.csproj" />
+    <ProjectReference Include="..\..\..\src\Microsoft.Agents.AI.Workflows\Microsoft.Agents.AI.Workflows.csproj" />
+  </ItemGroup>
+</Project>
@@ -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<List<string>> RunOrchestratorAsync(
+        [OrchestrationTrigger] TaskOrchestrationContext context)
+    {
+        ILogger logger = context.CreateReplaySafeLogger(nameof(MyOrchFunction));
+        logger.LogInformation("Saying hello.");
+        var outputs = new List<string>();
+
+        // Replace name and input with values relevant for your Durable Functions Activity
+        outputs.Add(await context.CallActivityAsync<string>(nameof(SayHello), "Tokyo"));
+        outputs.Add(await context.CallActivityAsync<string>(nameof(SayHello), "Seattle"));
+        outputs.Add(await context.CallActivityAsync<string>(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<HttpResponseData> 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);
+    }
+}
@@ -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;
+
+/// <summary>
+/// Constants for shared state scopes used across executors.
+/// </summary>
+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; }
+}
+
+/// <summary>
+/// First executor that processes a message and stores the result in shared state.
+/// </summary>
+internal sealed class OrderIdParserExecutor() : Executor<string, Order>("OrderIdParserExecutor")
+{
+    public override async ValueTask<Order> 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);
+    }
+}
+
+/// <summary>
+/// Second executor that reads the shared state and appends to the message.
+/// </summary>
+internal sealed class EmailSenderExecutor() : Executor<Order, string>("EmailSenderExecutor")
+{
+    public override async ValueTask<string> HandleAsync(Order message, IWorkflowContext context, CancellationToken cancellationToken = default)
+    {
+        // Read the processed message from shared state (written by OrderIdParserExecutor)
+        string? storedMessage = await context.ReadStateAsync<string>(
+            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<Order, Order>("PaymentProcesserExecutor")
+{
+    public override async ValueTask<Order> HandleAsync(Order message, IWorkflowContext context, CancellationToken cancellationToken = default)
+    {
+        // Call payment gateway.
+        message.PaymentReferenceNumber = Guid.NewGuid().ToString().Substring(0, 4);
+        return message;
+    }
+}
@@ -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();
@@ -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
+    }
+  }
+}
+```
@@ -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
@@ -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"
+      }
+    }
+  }
+}
@@ -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<string>(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;
        }

@@ -33,15 +33,17 @@ internal static class BuiltInFunctions
    // Exposed as an activity trigger for workflow executors
    public static Task<string> 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<DurableWorkflowRunner>();
-        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}");
@@ -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;
+
+/// <summary>
+/// An implementation of <see cref="IWorkflowContext"/> 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.
+/// </summary>
+/// <remarks>
+/// 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.
+/// </remarks>
+[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<string, string?> _pendingUpdates = [];
+    private readonly HashSet<string> _clearedScopes = [];
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="DurableExecutorContext"/> class.
+    /// </summary>
+    /// <param name="instanceId">The orchestration instance ID used to scope the state entity.</param>
+    /// <param name="client">The durable task client for entity operations.</param>
+    public DurableExecutorContext(string instanceId, DurableTaskClient client)
+    {
+        this._instanceId = instanceId;
+        this._client = client;
+    }
+
+    /// <inheritdoc/>
+    public ValueTask AddEventAsync(WorkflowEvent workflowEvent, CancellationToken cancellationToken = default)
+    {
+        // In activity context, events are not propagated to the workflow
+        return default;
+    }
+
+    /// <inheritdoc/>
+    public ValueTask SendMessageAsync(object message, string? targetId = null, CancellationToken cancellationToken = default)
+    {
+        // In activity context, messages cannot be routed to other executors
+        return default;
+    }
+
+    /// <inheritdoc/>
+    public ValueTask YieldOutputAsync(object output, CancellationToken cancellationToken = default)
+    {
+        // In activity context, outputs are not yielded to the workflow
+        return default;
+    }
+
+    /// <inheritdoc/>
+    public ValueTask RequestHaltAsync()
+    {
+        // Halt requests are not supported in activity context
+        return default;
+    }
+
+    /// <inheritdoc/>
+    public async ValueTask<T?> ReadStateAsync<T>(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<T>(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<WorkflowStateData>();
+        if (stateData?.Values is null)
+        {
+            return default;
+        }
+
+        if (stateData.Values.TryGetValue(scopeKey, out string? serializedValue) && serializedValue is not null)
+        {
+            return JsonSerializer.Deserialize<T>(serializedValue);
+        }
+
+        return default;
+    }
+
+    /// <inheritdoc/>
+    public async ValueTask<T> ReadOrInitStateAsync<T>(string key, Func<T> initialStateFactory, string? scopeName = null, CancellationToken cancellationToken = default)
+    {
+        T? value = await this.ReadStateAsync<T>(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;
+    }
+
+    /// <inheritdoc/>
+    public async ValueTask<HashSet<string>> ReadStateKeysAsync(string? scopeName = null, CancellationToken cancellationToken = default)
+    {
+        string normalizedScope = scopeName ?? "__default__";
+        string scopePrefix = GetScopePrefix(scopeName);
+        HashSet<string> 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<WorkflowStateData>();
+            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<string, string?> 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;
+    }
+
+    /// <inheritdoc/>
+    public async ValueTask QueueStateUpdateAsync<T>(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);
+    }
+
+    /// <inheritdoc/>
+    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<string> 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);
+    }
+
+    /// <inheritdoc/>
+    public IReadOnlyDictionary<string, string>? TraceContext => null;
+
+    /// <inheritdoc/>
+    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<string> GetPendingKeysForScope(string? scopeName)
+    {
+        string scopePrefix = GetScopePrefix(scopeName);
+        HashSet<string> keys = [];
+
+        foreach (KeyValuePair<string, string?> 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}:";
+    }
+}
@@ -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>(WorkflowSharedStateEntity.EntityName);
+
            foreach (string workflowName in workflows.Workflows.Select(kp => kp.Key))
            {
                tasks.AddOrchestratorFunc<DuableWorkflowRunRequest, List<string>>(
@@ -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,
@@ -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];
    }

+    /// <summary>
+    /// Cleans up the workflow state entity by signaling it to delete itself.
+    /// </summary>
+    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);
+    }
+
    /// <summary>
    /// Executes an activity function for a workflow executor.
    /// </summary>
    /// <param name="activityFunctionName">The name of the activity function to execute.</param>
-    /// <param name="input">The input string for the executor.</param>
-    /// <param name="functionContext">The Azure Functions context.</param>
-    /// <returns>The serialized result of the executor.</returns>
+    /// <param name="input">The serialized executor input.</param>
+    /// <param name="durableTaskClient">The durable task client for entity operations.</param>
+    /// <param name="functionContext">The function context containing binding data with the orchestration instance ID.</param>
+    /// <returns>The serialized executor output.</returns>
    internal async Task<string> 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<string> ExecuteWorkflowLevelsAsync(
        TaskOrchestrationContext context,
        Workflow workflow,
@@ -1,90 +0,0 @@
-// Copyright (c) Microsoft. All rights reserved.
-
-using Microsoft.Agents.AI.Workflows;
-
-namespace Microsoft.Agents.AI.Hosting.AzureFunctions;
-
-/// <summary>
-/// A minimal implementation of <see cref="IWorkflowContext"/> for use in Azure Functions activities.
-/// This provides basic context support for simple executors that don't require full workflow infrastructure.
-/// </summary>
-internal sealed class MinimalActivityContext : IWorkflowContext
-{
-    public MinimalActivityContext(string executorId)
-    {
-        // executorId is provided but not stored since this minimal context doesn't use it
-        _ = executorId;
-    }
-
-    /// <inheritdoc/>
-    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;
-    }
-
-    /// <inheritdoc/>
-    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;
-    }
-
-    /// <inheritdoc/>
-    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;
-    }
-
-    /// <inheritdoc/>
-    public ValueTask RequestHaltAsync()
-    {
-        // Halt requests are not supported in activity context
-        return default;
-    }
-
-    /// <inheritdoc/>
-    public ValueTask<T?> ReadStateAsync<T>(string key, string? scopeName = null, CancellationToken cancellationToken = default)
-    {
-        // No state available in activity context
-        return new ValueTask<T?>(default(T));
-    }
-
-    /// <inheritdoc/>
-    public ValueTask<T> ReadOrInitStateAsync<T>(string key, Func<T> initialStateFactory, string? scopeName = null, CancellationToken cancellationToken = default)
-    {
-        // Initialize with factory value since no state is available
-        return new ValueTask<T>(initialStateFactory());
-    }
-
-    /// <inheritdoc/>
-    public ValueTask<HashSet<string>> ReadStateKeysAsync(string? scopeName = null, CancellationToken cancellationToken = default)
-    {
-        // No state keys in activity context
-        return new ValueTask<HashSet<string>>([]);
-    }
-
-    /// <inheritdoc/>
-    public ValueTask QueueStateUpdateAsync<T>(string key, T? value, string? scopeName = null, CancellationToken cancellationToken = default)
-    {
-        // State updates are not persisted in activity context
-        return default;
-    }
-
-    /// <inheritdoc/>
-    public ValueTask QueueClearScopeAsync(string? scopeName = null, CancellationToken cancellationToken = default)
-    {
-        // No state to clear in activity context
-        return default;
-    }
-
-    /// <inheritdoc/>
-    public IReadOnlyDictionary<string, string>? TraceContext => null;
-
-    /// <inheritdoc/>
-    public bool ConcurrentRunsEnabled => false;
-}
@@ -0,0 +1,171 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using Microsoft.DurableTask.Entities;
+
+namespace Microsoft.Agents.AI.Hosting.AzureFunctions;
+
+/// <summary>
+/// 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.
+/// </summary>
+internal sealed class WorkflowSharedStateEntity : TaskEntity<WorkflowStateData>
+{
+    /// <summary>
+    /// The entity name used for registration and lookup.
+    /// </summary>
+    public const string EntityName = "workflow-shared-state";
+
+    /// <summary>
+    /// Reads a state value by key and scope.
+    /// </summary>
+    /// <param name="request">The read request containing key and optional scope.</param>
+    /// <returns>The serialized state value, or null if not found.</returns>
+    public string? ReadState(WorkflowStateReadRequest request)
+    {
+        string scopeKey = GetScopeKey(request.ScopeName, request.Key);
+        return this.State.Values.TryGetValue(scopeKey, out string? value) ? value : null;
+    }
+
+    /// <summary>
+    /// Reads the entire state dictionary.
+    /// </summary>
+    /// <returns>A copy of the current state.</returns>
+    public Dictionary<string, string> ReadAllState()
+    {
+        return new Dictionary<string, string>(this.State.Values);
+    }
+
+    /// <summary>
+    /// Writes or updates a state value by key and scope.
+    /// </summary>
+    /// <param name="request">The write request containing key, scope, and value.</param>
+    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;
+        }
+    }
+
+    /// <summary>
+    /// Gets all keys within a specific scope.
+    /// </summary>
+    /// <param name="scopeName">The scope name, or null for the default scope.</param>
+    /// <returns>A collection of keys within the scope.</returns>
+    public HashSet<string> GetStateKeys(string? scopeName)
+    {
+        string scopePrefix = GetScopePrefix(scopeName);
+        HashSet<string> 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;
+    }
+
+    /// <summary>
+    /// Clears all state entries within a specific scope.
+    /// </summary>
+    /// <param name="scopeName">The scope name, or null for the default scope.</param>
+    public void ClearScope(string? scopeName)
+    {
+        string scopePrefix = GetScopePrefix(scopeName);
+        List<string> 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);
+        }
+    }
+
+    /// <summary>
+    /// Deletes the entity, cleaning up all state.
+    /// Called by the orchestration when it completes.
+    /// </summary>
+    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}:";
+    }
+}
+
+/// <summary>
+/// Represents the internal state data for a workflow state entity.
+/// </summary>
+internal sealed class WorkflowStateData
+{
+    /// <summary>
+    /// Gets the state dictionary mapping scope-prefixed keys to serialized values.
+    /// </summary>
+    public Dictionary<string, string> Values { get; init; } = [];
+}
+
+/// <summary>
+/// Request model for reading workflow state.
+/// </summary>
+internal sealed class WorkflowStateReadRequest
+{
+    /// <summary>
+    /// Gets or sets the state key.
+    /// </summary>
+    public string Key { get; set; } = string.Empty;
+
+    /// <summary>
+    /// Gets or sets the optional scope name.
+    /// </summary>
+    public string? ScopeName { get; set; }
+}
+
+/// <summary>
+/// Request model for writing workflow state.
+/// </summary>
+internal sealed class WorkflowStateWriteRequest
+{
+    /// <summary>
+    /// Gets or sets the state key.
+    /// </summary>
+    public string Key { get; set; } = string.Empty;
+
+    /// <summary>
+    /// Gets or sets the optional scope name.
+    /// </summary>
+    public string? ScopeName { get; set; }
+
+    /// <summary>
+    /// Gets or sets the serialized value, or null to delete the key.
+    /// </summary>
+    public string? Value { get; set; }
+}