diff --git a/dotnet/agent-framework-dotnet.slnx b/dotnet/agent-framework-dotnet.slnx index c719adfa6e..bbd71d6fba 100644 --- a/dotnet/agent-framework-dotnet.slnx +++ b/dotnet/agent-framework-dotnet.slnx @@ -50,6 +50,10 @@ + + + + @@ -417,8 +421,8 @@ - + diff --git a/dotnet/samples/AzureFunctions/10_WorkflowConcurrent/Program.cs b/dotnet/samples/AzureFunctions/10_WorkflowConcurrent/Program.cs index a1435fd39d..78236c18f1 100644 --- a/dotnet/samples/AzureFunctions/10_WorkflowConcurrent/Program.cs +++ b/dotnet/samples/AzureFunctions/10_WorkflowConcurrent/Program.cs @@ -24,8 +24,8 @@ AzureOpenAIClient client = !string.IsNullOrEmpty(azureOpenAiKey) ? new AzureOpenAIClient(new Uri(endpoint), new AzureKeyCredential(azureOpenAiKey)) : new AzureOpenAIClient(new Uri(endpoint), new AzureCliCredential()); -AIAgent physicist = client.GetChatClient(deploymentName).CreateAIAgent("You are an expert in physics. You answer questions from a physics perspective.", "Physicist"); -AIAgent chemist = client.GetChatClient(deploymentName).CreateAIAgent("You are an expert in chemistry. You answer questions from a chemistry perspective.", "Chemist"); +AIAgent physicist = client.GetChatClient(deploymentName).AsAIAgent("You are an expert in physics. You answer questions from a physics perspective.", "Physicist"); +AIAgent chemist = client.GetChatClient(deploymentName).AsAIAgent("You are an expert in chemistry. You answer questions from a chemistry perspective.", "Chemist"); var startExecutor = new ConcurrentStartExecutor(); var aggregationExecutor = new ResultAggregationExecutor(); diff --git a/dotnet/samples/DurableAgents/ConsoleApps/08_SingleWorkflow/08_SingleWorkflow.csproj b/dotnet/samples/DurableAgents/ConsoleApps/08_SingleWorkflow/08_SingleWorkflow.csproj new file mode 100644 index 0000000000..81439ef42b --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/08_SingleWorkflow/08_SingleWorkflow.csproj @@ -0,0 +1,29 @@ + + + net10.0 + Exe + enable + enable + SingleWorkflow + SingleAgent + + + + + + + + + + + + + + + + diff --git a/dotnet/samples/DurableAgents/ConsoleApps/08_SingleWorkflow/OrderCancelExecutor.cs b/dotnet/samples/DurableAgents/ConsoleApps/08_SingleWorkflow/OrderCancelExecutor.cs new file mode 100644 index 0000000000..153830c84b --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/08_SingleWorkflow/OrderCancelExecutor.cs @@ -0,0 +1,128 @@ +// Copyright (c) Microsoft. All rights reserved. + +using Microsoft.Agents.AI.Workflows; + +namespace SingleAgent; + +/// +/// Represents an order in the system. +/// +internal sealed class Order +{ + public required string Id { get; set; } + public DateTime OrderDate { get; set; } + public bool IsCancelled { get; set; } + public required Customer Customer { get; set; } +} + +/// +/// Represents a customer associated with an order. +/// +internal sealed class Customer +{ + public string Name { get; set; } = string.Empty; + public string Email { get; set; } = string.Empty; +} + +/// +/// Looks up an order by its ID. +/// This activity simulates a database lookup with a 2 second delay. +/// +internal sealed class OrderLookup() : Executor("OrderLookup") +{ + public override async ValueTask HandleAsync(string message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + // Log that this activity is executing (not replaying) + Console.WriteLine(); + Console.ForegroundColor = ConsoleColor.Magenta; + Console.WriteLine("┌─────────────────────────────────────────────────────────────────┐"); + Console.WriteLine($"│ [Activity] OrderLookup: Starting lookup for order '{message}'"); + Console.ResetColor(); + + // Simulate database lookup with delay + await Task.Delay(TimeSpan.FromSeconds(2), cancellationToken); + + Order order = new() + { + Id = message, + OrderDate = DateTime.UtcNow.AddDays(-3), + IsCancelled = false, + Customer = new Customer { Name = "Jerry", Email = "jerry@example.com" } + }; + + Console.ForegroundColor = ConsoleColor.Magenta; + Console.WriteLine($"│ [Activity] OrderLookup: Found order '{message}' for customer '{order.Customer.Name}'"); + Console.WriteLine("└─────────────────────────────────────────────────────────────────┘"); + Console.ResetColor(); + + return order; + } +} + +/// +/// Cancels an order. +/// This activity simulates a slow cancellation process with a 5 second delay. +/// Try pressing Ctrl+C during this activity to see durability in action! +/// +internal sealed class OrderCancel() : Executor("OrderCancel") +{ + public override async ValueTask HandleAsync(Order message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + // Log that this activity is executing (not replaying) + Console.WriteLine(); + Console.ForegroundColor = ConsoleColor.Yellow; + Console.WriteLine("┌─────────────────────────────────────────────────────────────────┐"); + Console.WriteLine($"│ [Activity] OrderCancel: Starting cancellation for order '{message.Id}'"); + Console.WriteLine("│ [Activity] OrderCancel: ⚠️ This takes 5 seconds - try Ctrl+C!"); + Console.ResetColor(); + + // Simulate a slow cancellation process (e.g., calling external payment system) + // This is where you can kill the process to test durability + for (int i = 1; i <= 5; i++) + { + await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken); + Console.ForegroundColor = ConsoleColor.DarkYellow; + Console.WriteLine($"│ [Activity] OrderCancel: Processing... {i}/5 seconds"); + Console.ResetColor(); + } + + // Mark the order as cancelled + message.IsCancelled = true; + + Console.ForegroundColor = ConsoleColor.Yellow; + Console.WriteLine($"│ [Activity] OrderCancel: ✓ Order '{message.Id}' has been cancelled"); + Console.WriteLine("└─────────────────────────────────────────────────────────────────┘"); + Console.ResetColor(); + + return message; + } +} + +/// +/// Sends a cancellation confirmation email to the customer. +/// This activity simulates sending an email with a 1 second delay. +/// +internal sealed class SendEmail() : Executor("SendEmail") +{ + public override async ValueTask HandleAsync(Order message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + // Log that this activity is executing (not replaying) + Console.WriteLine(); + Console.ForegroundColor = ConsoleColor.Cyan; + Console.WriteLine("┌─────────────────────────────────────────────────────────────────┐"); + Console.WriteLine($"│ [Activity] SendEmail: Sending email to '{message.Customer.Email}'..."); + Console.ResetColor(); + + // Simulate email sending delay + await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken); + + string result = $"Cancellation email sent to {message.Customer.Email} for order {message.Id}."; + + Console.ForegroundColor = ConsoleColor.Cyan; + Console.WriteLine("│ [Activity] SendEmail: ✓ Email sent successfully!"); + Console.WriteLine("└─────────────────────────────────────────────────────────────────┘"); + Console.ResetColor(); + + return result; + } +} diff --git a/dotnet/samples/DurableAgents/ConsoleApps/08_SingleWorkflow/Program.cs b/dotnet/samples/DurableAgents/ConsoleApps/08_SingleWorkflow/Program.cs new file mode 100644 index 0000000000..0ec1b404d5 --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/08_SingleWorkflow/Program.cs @@ -0,0 +1,107 @@ +// Copyright (c) Microsoft. All rights reserved. + +// This sample demonstrates how to run a workflow as a durable orchestration from a console application. +// The workflow consists of three executors: OrderLookup -> OrderCancel -> SendEmail. +// It uses the DurableExecution API similar to InProcessExecution for in-process workflows. +// +// DURABILITY DEMONSTRATION: +// - Each activity has artificial delays to simulate real-world operations +// - Stop the app (Ctrl+C or stop debugging) during the OrderCancel activity (5 seconds) +// - Restart the application - the workflow will automatically resume! +// - The Durable Task Framework will skip already-completed activities (OrderLookup) +// and continue from where it left off (OrderCancel) + +using Microsoft.Agents.AI.DurableTask; +using Microsoft.Agents.AI.Workflows; +using Microsoft.DurableTask.Client; +using Microsoft.DurableTask.Client.AzureManaged; +using Microsoft.DurableTask.Worker.AzureManaged; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using SingleAgent; + +// Get DTS connection string from environment variable +string dtsConnectionString = Environment.GetEnvironmentVariable("DURABLE_TASK_SCHEDULER_CONNECTION_STRING") + ?? "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None"; + +// Define executors for the workflow +OrderLookup orderLookup = new(); +OrderCancel orderCancel = new(); +SendEmail sendEmail = new(); + +// Build the CancelOrder workflow: OrderLookup -> OrderCancel -> SendEmail +Workflow cancelOrder = new WorkflowBuilder(orderLookup) + .WithName("CancelOrder") + .WithDescription("Cancel an order and notify the customer") + .AddEdge(orderLookup, orderCancel) + .AddEdge(orderCancel, sendEmail) + .Build(); + +IHost host = Host.CreateDefaultBuilder(args) + .ConfigureLogging(logging => logging.SetMinimumLevel(LogLevel.Warning)) + .ConfigureServices(services => + { + services.ConfigureDurableWorkflows( + options => options.Workflows.AddWorkflow(cancelOrder), + workerBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString), + clientBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString)); + }) + .Build(); + +await host.StartAsync(); + +DurableTaskClient durableClient = host.Services.GetRequiredService(); + +Console.WriteLine("Durable Workflow Sample"); +Console.WriteLine("Workflow: OrderLookup (2s) -> OrderCancel (5s) -> SendEmail (1s)"); +Console.WriteLine(); +Console.WriteLine("TIP: Stop the app during OrderCancel to test durability."); +Console.WriteLine(" Restart - it will resume from where it left off."); +Console.WriteLine(); +Console.WriteLine("Checking for pending workflows..."); +await Task.Delay(TimeSpan.FromSeconds(2)); +Console.WriteLine(); +Console.WriteLine("Enter an order ID (or 'exit'):"); + +while (true) +{ + Console.Write("> "); + string? input = Console.ReadLine(); + if (string.IsNullOrWhiteSpace(input) || input.Equals("exit", StringComparison.OrdinalIgnoreCase)) + { + break; + } + + try + { + await StartNewWorkflowAsync(input, cancelOrder, durableClient); + } + catch (Exception ex) + { + Console.WriteLine($"Error: {ex.Message}"); + } + + Console.WriteLine(); +} + +await host.StopAsync(); + +// Start a new workflow +async Task StartNewWorkflowAsync(string orderId, Workflow workflow, DurableTaskClient client) +{ + Console.WriteLine($"Starting workflow for order '{orderId}'..."); + + await using DurableRun run = await DurableWorkflow.RunAsync(workflow, orderId, client); + Console.WriteLine($"Instance ID: {run.InstanceId}"); + + try + { + string? result = await run.WaitForCompletionAsync(); + Console.WriteLine($"Completed: {result}"); + } + catch (InvalidOperationException ex) + { + Console.WriteLine($"Failed: {ex.Message}"); + } +} diff --git a/dotnet/samples/DurableAgents/ConsoleApps/08_SingleWorkflow/README.md b/dotnet/samples/DurableAgents/ConsoleApps/08_SingleWorkflow/README.md new file mode 100644 index 0000000000..0ea6210826 --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/08_SingleWorkflow/README.md @@ -0,0 +1,156 @@ +# Single Workflow Console Sample + +This sample demonstrates how to run a workflow as a durable orchestration from a console application using the Durable Task Framework. It showcases the **durability** aspect - if the process crashes mid-execution, the workflow can be resumed without re-executing completed activities. + +## Overview + +The sample implements an order cancellation workflow with three executors, each with artificial delays to simulate real-world operations: + +1. **OrderLookup** (2 seconds) - Looks up an order by its ID +2. **OrderCancel** (5 seconds) - Marks the order as cancelled +3. **SendEmail** (1 second) - Sends a cancellation confirmation email + +## Durability Demonstration + +The key feature of Durable Task Framework is **durability**: + + +- **Activity results are persisted**: When an activity completes, its result is saved +- **Orchestrations are replayed**: On restart, the orchestration replays from the beginning +- **Completed activities are skipped**: The framework uses cached results for completed activities +- **Failed activities are retried**: If an activity was interrupted, it runs again +- **Automatic resume**: When the worker starts, it automatically picks up any pending work! + +### Try It Yourself + +1. Start the application and enter an order ID (e.g., `12345`) +2. Stop the app (Ctrl+C or stop debugging) during the `OrderCancel` activity (5 seconds) +3. Restart the application +4. **Watch for automatic resume!** The worker automatically picks up the interrupted workflow +5. Observe that `OrderLookup` is NOT re-executed (its result was cached) +6. `OrderCancel` restarts from the beginning (it didn't complete) +7. `SendEmail` runs after `OrderCancel` completes + +The durability is completely automatic - no manual intervention needed! + +## Workflow Flow + +``` +User Input (Order ID) + ? + ? +??????????????????? +? OrderLookup ? ? 2 second delay (database lookup) +? (2 seconds) ? +??????????????????? + ? + ? +??????????????????? +? OrderCancel ? ? 5 second delay - TRY INTERRUPTING HERE! +? (5 seconds) ? +??????????????????? + ? + ? +??????????????????? +? SendEmail ? ? 1 second delay (email sending) +? (1 second) ? +??????????????????? + ? + ? + Result +``` + +## Key Concepts Demonstrated + +- **ConfigureDurableWorkflows** - Simplified API for registering workflows +- **DurableExecution.RunAsync** - Start a new workflow (similar to InProcessExecution) +- **DurableRun** - Handle to monitor and interact with a running workflow +- **Automatic Resume** - Interrupted workflows continue automatically on restart + +## 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, you can run the sample: + +```bash +cd dotnet/samples/DurableAgents/ConsoleApps/08_SingleWorkflow +dotnet run --framework net10.0 +``` + +### Sample Session + +```text +?????????????????????????????????????????????????????????????????????? +? Durable Workflow Console Sample ? +?????????????????????????????????????????????????????????????????????? +? This sample demonstrates durability in workflows. ? +? Workflow: OrderLookup (2s) -> OrderCancel (5s) -> SendEmail (1s) ? +?????????????????????????????????????????????????????????????????????? + +?? TIP: Stop the app during OrderCancel (5 seconds) to test durability! + Restart the app - it will automatically resume from where it left off. + +? Checking for pending workflows... + +Enter an order ID to start a new workflow (or 'exit' to quit): + +Order ID: 12345 + +Starting workflow for order '12345'... +Instance ID: abc123-def456-... + +??????????????????????????????????????????????????????????????????? +? [Activity] OrderLookup: Starting lookup for order '12345' +? [Activity] OrderLookup: Found order '12345' for customer 'Jerry' +??????????????????????????????????????????????????????????????????? + +??????????????????????????????????????????????????????????????????? +? [Activity] OrderCancel: Starting cancellation for order '12345' +? [Activity] OrderCancel: ?? This takes 5 seconds - try Ctrl+C! +? [Activity] OrderCancel: Processing... 1/5 seconds +? [Activity] OrderCancel: Processing... 2/5 seconds +^C <-- User stops the app here + +[After restart...] + +? Checking for pending workflows... + +??????????????????????????????????????????????????????????????????? +? [Activity] OrderCancel: Starting cancellation for order '12345' <-- Auto-resumed! +? [Activity] OrderCancel: ?? This takes 5 seconds - try Ctrl+C! +? [Activity] OrderCancel: Processing... 1/5 seconds +... +? [Activity] OrderCancel: ? Order '12345' has been cancelled +??????????????????????????????????????????????????????????????????? + +??????????????????????????????????????????????????????????????????? +? [Activity] SendEmail: Sending email to 'jerry@example.com'... +? [Activity] SendEmail: ? Email sent successfully! +??????????????????????????????????????????????????????????????????? + +Enter an order ID to start a new workflow (or 'exit' to quit): + +Order ID: _ +``` + +Notice that when resumed: +- `OrderLookup` was **NOT re-executed** (result was cached by Durable Task) +- `OrderCancel` **restarted automatically** (it was interrupted before completing) +- `SendEmail` ran normally after `OrderCancel` completed + +## Viewing Workflow State + +You can view the state of the workflow in the Durable Task Scheduler dashboard: + +1. Open your browser and navigate to `http://localhost:8082` +2. In the dashboard, you can view the state of the orchestration, including activity history and current state + +## Related Samples + +- [01_SingleAgent](../01_SingleAgent) - Single agent console sample +- [02_AgentOrchestration_Chaining](../02_AgentOrchestration_Chaining) - Agent chaining with durable orchestration +- [05_AgentOrchestration_HITL](../05_AgentOrchestration_HITL) - Human-in-the-loop orchestration +- [09_Workflow](../../AzureFunctions/09_Workflow) - Azure Functions version of workflow hosting diff --git a/dotnet/samples/DurableAgents/ConsoleApps/09_Workflow_Concurrency/09_Workflow_Concurrency.csproj b/dotnet/samples/DurableAgents/ConsoleApps/09_Workflow_Concurrency/09_Workflow_Concurrency.csproj new file mode 100644 index 0000000000..eb31eda874 --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/09_Workflow_Concurrency/09_Workflow_Concurrency.csproj @@ -0,0 +1,31 @@ + + + net10.0 + Exe + enable + enable + SingleWorkflow + SingleAgent + + + + + + + + + + + + + + + + + + diff --git a/dotnet/samples/DurableAgents/ConsoleApps/09_Workflow_Concurrency/ExpertExecutors.cs b/dotnet/samples/DurableAgents/ConsoleApps/09_Workflow_Concurrency/ExpertExecutors.cs new file mode 100644 index 0000000000..50ea489bd4 --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/09_Workflow_Concurrency/ExpertExecutors.cs @@ -0,0 +1,67 @@ +// Copyright (c) Microsoft. All rights reserved. + +using Microsoft.Agents.AI.Workflows; + +namespace WorkflowConcurrency; + +/// +/// Parses and validates the incoming question before sending to AI agents. +/// +internal sealed class ParseQuestionExecutor() : Executor("ParseQuestion") +{ + public override ValueTask HandleAsync(string message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + Console.WriteLine(); + Console.ForegroundColor = ConsoleColor.Magenta; + Console.WriteLine("┌─────────────────────────────────────────────────────────────────┐"); + Console.WriteLine("│ [ParseQuestion] Preparing question for AI agents..."); + + string formattedQuestion = message.Trim(); + if (!formattedQuestion.EndsWith('?')) + { + formattedQuestion += "?"; + } + + Console.WriteLine($"│ [ParseQuestion] Question: \"{formattedQuestion}\""); + Console.WriteLine("│ [ParseQuestion] → Sending to Physicist and Chemist in PARALLEL..."); + Console.WriteLine("└─────────────────────────────────────────────────────────────────┘"); + Console.ResetColor(); + + return ValueTask.FromResult(formattedQuestion); + } +} + +/// +/// Aggregates responses from all AI agents into a comprehensive answer. +/// This is the Fan-in point where parallel results are collected. +/// +internal sealed class AggregatorExecutor() : Executor("Aggregator") +{ + public override ValueTask HandleAsync(string[] message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + Console.WriteLine(); + Console.ForegroundColor = ConsoleColor.Cyan; + Console.WriteLine("┌─────────────────────────────────────────────────────────────────┐"); + Console.WriteLine($"│ [Aggregator] 📋 Received {message.Length} AI agent responses"); + Console.WriteLine("│ [Aggregator] Combining into comprehensive answer..."); + Console.WriteLine("│ [Aggregator] ✓ Aggregation complete!"); + Console.WriteLine("└─────────────────────────────────────────────────────────────────┘"); + Console.ResetColor(); + + string aggregatedResult = "═══════════════════════════════════════════════════════════════\n" + + " AI EXPERT PANEL RESPONSES\n" + + "═══════════════════════════════════════════════════════════════\n\n"; + + for (int i = 0; i < message.Length; i++) + { + string expertLabel = i == 0 ? "⚛️ PHYSICIST" : "🧪 CHEMIST"; + aggregatedResult += $"{expertLabel}:\n{message[i]}\n\n"; + } + + aggregatedResult += "═══════════════════════════════════════════════════════════════\n" + + $"Summary: Received perspectives from {message.Length} AI experts.\n" + + "═══════════════════════════════════════════════════════════════"; + + return ValueTask.FromResult(aggregatedResult); + } +} diff --git a/dotnet/samples/DurableAgents/ConsoleApps/09_Workflow_Concurrency/OrderCancelExecutor.cs b/dotnet/samples/DurableAgents/ConsoleApps/09_Workflow_Concurrency/OrderCancelExecutor.cs new file mode 100644 index 0000000000..e69de29bb2 diff --git a/dotnet/samples/DurableAgents/ConsoleApps/09_Workflow_Concurrency/Program.cs b/dotnet/samples/DurableAgents/ConsoleApps/09_Workflow_Concurrency/Program.cs new file mode 100644 index 0000000000..528d709e41 --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/09_Workflow_Concurrency/Program.cs @@ -0,0 +1,134 @@ +// Copyright (c) Microsoft. All rights reserved. + +// This sample demonstrates the Fan-out/Fan-in pattern in a durable workflow. +// The workflow uses 4 executors: 2 class-based executors and 2 AI agents. +// +// WORKFLOW PATTERN (4 Executors): +// +// ┌──────────────────┐ +// │ ParseQuestion │ ← Class-based Executor +// └────────┬─────────┘ +// │ +// ┌────────┴────────┐ +// ▼ ▼ +// ┌──────────┐ ┌──────────┐ +// │ Physicist│ │ Chemist │ ← AI Agents (parallel) +// └────┬─────┘ └────┬─────┘ +// │ │ +// └──────┬───────┘ +// ▼ +// ┌──────────────────┐ +// │ Aggregator │ ← Class-based Executor +// └──────────────────┘ + +using Azure; +using Azure.AI.OpenAI; +using Azure.Identity; +using Microsoft.Agents.AI; +using Microsoft.Agents.AI.DurableTask; +using Microsoft.Agents.AI.Workflows; +using Microsoft.DurableTask.Client; +using Microsoft.DurableTask.Client.AzureManaged; +using Microsoft.DurableTask.Worker.AzureManaged; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using OpenAI.Chat; +using WorkflowConcurrency; + +// Configuration +string dtsConnectionString = Environment.GetEnvironmentVariable("DURABLE_TASK_SCHEDULER_CONNECTION_STRING") + ?? "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None"; +string endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT") + ?? throw new InvalidOperationException("AZURE_OPENAI_ENDPOINT is not set."); +string deploymentName = Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT") + ?? throw new InvalidOperationException("AZURE_OPENAI_DEPLOYMENT is not set."); +string? azureOpenAiKey = Environment.GetEnvironmentVariable("AZURE_OPENAI_KEY"); + +// Create Azure OpenAI client +AzureOpenAIClient openAiClient = !string.IsNullOrEmpty(azureOpenAiKey) + ? new AzureOpenAIClient(new Uri(endpoint), new AzureKeyCredential(azureOpenAiKey)) + : new AzureOpenAIClient(new Uri(endpoint), new AzureCliCredential()); +ChatClient chatClient = openAiClient.GetChatClient(deploymentName); + +// Define the 4 executors for the workflow +ParseQuestionExecutor parseQuestion = new(); // Executor 1: Class-based +AIAgent physicist = chatClient.AsAIAgent("You are a physics expert. Be concise (2-3 sentences).", "Physicist"); // Executor 2: AI Agent +AIAgent chemist = chatClient.AsAIAgent("You are a chemistry expert. Be concise (2-3 sentences).", "Chemist"); // Executor 3: AI Agent +AggregatorExecutor aggregator = new(); // Executor 4: Class-based + +// Build workflow: ParseQuestion → [Physicist, Chemist] (parallel) → Aggregator +Workflow workflow = new WorkflowBuilder(parseQuestion) + .WithName("ExpertReview") + .AddFanOutEdge(parseQuestion, [physicist, chemist]) + .AddFanInEdge([physicist, chemist], aggregator) + .Build(); + +// Configure and start the host +IHost host = Host.CreateDefaultBuilder(args) + .ConfigureLogging(logging => logging.SetMinimumLevel(LogLevel.Warning)) + .ConfigureServices(services => + { + services.ConfigureDurableWorkflows( + options => options.Workflows.AddWorkflow(workflow), + workerBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString), + clientBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString)); + }) + .Build(); + +await host.StartAsync(); +DurableTaskClient durableClient = host.Services.GetRequiredService(); + +// Console UI +Console.ForegroundColor = ConsoleColor.Cyan; +Console.WriteLine("╔═══════════════════════════════════════════════════════════════════════╗"); +Console.WriteLine("║ Fan-out/Fan-in Workflow Sample (4 Executors) ║"); +Console.WriteLine("║ ║"); +Console.WriteLine("║ ParseQuestion → [Physicist, Chemist] → Aggregator ║"); +Console.WriteLine("║ (class-based) (AI agents, parallel) (class-based) ║"); +Console.WriteLine("╚═══════════════════════════════════════════════════════════════════════╝"); +Console.ResetColor(); +Console.WriteLine(); + +await Task.Delay(TimeSpan.FromSeconds(2)); // Allow pending workflows to resume + +Console.WriteLine("Enter a science question (or 'exit' to quit):"); +Console.WriteLine(); + +while (true) +{ + Console.ForegroundColor = ConsoleColor.Yellow; + Console.Write("Question: "); + Console.ResetColor(); + + string? input = Console.ReadLine(); + if (string.IsNullOrWhiteSpace(input) || input.Equals("exit", StringComparison.OrdinalIgnoreCase)) + { + break; + } + + try + { + await using DurableRun run = await DurableWorkflow.RunAsync(workflow, input, durableClient); + Console.ForegroundColor = ConsoleColor.Gray; + Console.WriteLine($"Instance: {run.InstanceId}"); + Console.ResetColor(); + + string? result = await run.WaitForCompletionAsync(); + + Console.ForegroundColor = ConsoleColor.Green; + Console.WriteLine("\n✓ Workflow completed!\n"); + Console.ResetColor(); + Console.WriteLine(result); + } + catch (Exception ex) + { + Console.ForegroundColor = ConsoleColor.Red; + Console.WriteLine($"✗ Error: {ex.Message}"); + Console.ResetColor(); + } + + Console.WriteLine(); +} + +await host.StopAsync(); diff --git a/dotnet/samples/DurableAgents/ConsoleApps/09_Workflow_Concurrency/README.md b/dotnet/samples/DurableAgents/ConsoleApps/09_Workflow_Concurrency/README.md new file mode 100644 index 0000000000..98c35e12ae --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/09_Workflow_Concurrency/README.md @@ -0,0 +1,160 @@ +# Fan-out/Fan-in Workflow with AI Agents + +This sample demonstrates the **Fan-out/Fan-in pattern** using real AI agents in a durable workflow. A question is sent to multiple AI "expert" agents in parallel, and their responses are aggregated into a final result. + +## Overview + +The sample implements an expert consultation workflow using Azure OpenAI: +- A question is parsed and prepared +- The question is sent to **2 AI agents in parallel** (Fan-out) +- AI agent responses are **collected and aggregated** (Fan-in) + +### Components + +| Component | Type | Description | +|-----------|------|-------------| +| ParseQuestion | Executor | Validates and formats the incoming question | +| Physicist | AI Agent | Azure OpenAI agent with physics expertise | +| Chemist | AI Agent | Azure OpenAI agent with chemistry expertise | +| Aggregator | Executor | Combines all AI agent responses | + +## Workflow Pattern + +``` + ??????????????????? + ? ParseQuestion ? + ??????????????????? + ? + ??????????????????????????????? + ? ? + ? ? + ??????????????????? ??????????????????? + ? Physicist ? ? Chemist ? + ? (AI Agent) ? ? (AI Agent) ? + ??????????????????? ??????????????????? + ? ? + ? PARALLEL EXECUTION ? + ? ? + ??????????????????????????????? + ? + ? + ??????????????????? + ? Aggregator ? + ??????????????????? +``` + +## Key Concepts Demonstrated + +- **Fan-out (AddFanOutEdge)**: One executor sends to multiple AI agents in parallel +- **Fan-in (AddFanInEdge)**: Multiple AI agent results are collected into an array +- **AI Agents in Workflows**: Using Azure OpenAI agents as workflow executors +- **Durability**: If interrupted, completed AI agent responses are preserved on restart + +## Configuration + +When using AI agents in durable workflows, `ConfigureDurableWorkflows` automatically registers any AI agents found in the workflow. You only need a single configuration call: + +```csharp +services.ConfigureDurableWorkflows( + options => options.Workflows.AddWorkflow(expertReview), + workerBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString), + clientBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString)); +``` + +This is similar to how `ConfigureDurableOptions` works in Azure Functions samples. + +## Environment Setup + +This sample requires: +1. **Durable Task Scheduler** - See the [parent README](../README.md) for setup instructions +2. **Azure OpenAI** - You need an Azure OpenAI resource with a deployed model + +### Environment Variables + +| Variable | Description | +|----------|-------------| +| `DURABLE_TASK_SCHEDULER_CONNECTION_STRING` | Connection string for DTS (defaults to local emulator) | +| `AZURE_OPENAI_ENDPOINT` | Your Azure OpenAI endpoint URL | +| `AZURE_OPENAI_DEPLOYMENT` | Name of your deployed model (e.g., `gpt-4`) | +| `AZURE_OPENAI_KEY` | (Optional) API key - if not set, uses Azure CLI credential | + +## Running the Sample + +```bash +cd dotnet/samples/DurableAgents/ConsoleApps/09_Workflow_Concurrency +dotnet run --framework net10.0 +``` + +### Sample Session + +```text +?????????????????????????????????????????????????????????????????????? +? Fan-out/Fan-in Workflow with AI Agents ? +?????????????????????????????????????????????????????????????????????? +? This sample demonstrates parallel AI agent consultation. ? +? ? +? Workflow: Question ? [Physicist, Chemist] ? Aggregator ? +? (AI agents run in parallel) ? +?????????????????????????????????????????????????????????????????????? + +?? TIP: Stop during execution to test durability - completed agent responses are preserved! + +? Checking for pending workflows... + +Enter a science question (or 'exit' to quit): + Example: "What is water?" + +Question: What is water + +Starting expert review workflow... +Instance ID: abc123-def456-... + +??????????????????????????????????????????????????????????????????? +? [ParseQuestion] Preparing question for AI agents... +? [ParseQuestion] Question: "What is water?" +? [ParseQuestion] ? Sending to Physicist and Chemist in PARALLEL... +??????????????????????????????????????????????????????????????????? + +... (AI agents process in parallel) + +??????????????????????????????????????????????????????????????????? +? [Aggregator] ?? Received 2 AI agent responses +? [Aggregator] Combining into comprehensive answer... +? [Aggregator] ? Aggregation complete! +??????????????????????????????????????????????????????????????????? + +?????????????????????????????????????????????????????????????????????? +? ? Expert review completed! ? +?????????????????????????????????????????????????????????????????????? + +??????????????????????????????????????????????????????????????? + AI EXPERT PANEL RESPONSES +??????????????????????????????????????????????????????????????? + +?? PHYSICIST: +Water is H2O - two hydrogen atoms bonded to one oxygen atom. From a physics +perspective, water exhibits unique properties like high specific heat capacity +and surface tension due to hydrogen bonding. + +?? CHEMIST: +Water (H2O) is a polar molecule formed by covalent bonds. Its bent molecular +geometry creates a dipole moment, making it an excellent solvent for ionic +and polar compounds. + +??????????????????????????????????????????????????????????????? +Summary: Received perspectives from 2 AI experts. +??????????????????????????????????????????????????????????????? +``` + +## Durability with AI Agents + +If you stop the application while one AI agent is processing: +- The completed agent's response is **preserved** (cached by Durable Task) +- On restart, only the incomplete agent **re-runs** +- The aggregator waits for all agents to complete + +## Related Samples + +- [08_SingleWorkflow](../08_SingleWorkflow) - Sequential workflow with durability demonstration +- [10_WorkflowConcurrent](../../AzureFunctions/10_WorkflowConcurrent) - Azure Functions version +- [02_AgentOrchestration_Chaining](../02_AgentOrchestration_Chaining) - Agent chaining with durable orchestration diff --git a/dotnet/samples/DurableAgents/ConsoleApps/10_Workflow_HITL/10_Workflow_HITL.csproj b/dotnet/samples/DurableAgents/ConsoleApps/10_Workflow_HITL/10_Workflow_HITL.csproj new file mode 100644 index 0000000000..81439ef42b --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/10_Workflow_HITL/10_Workflow_HITL.csproj @@ -0,0 +1,29 @@ + + + net10.0 + Exe + enable + enable + SingleWorkflow + SingleAgent + + + + + + + + + + + + + + + + diff --git a/dotnet/samples/DurableAgents/ConsoleApps/10_Workflow_HITL/CreateApprovalRequest.cs b/dotnet/samples/DurableAgents/ConsoleApps/10_Workflow_HITL/CreateApprovalRequest.cs new file mode 100644 index 0000000000..0393704e70 --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/10_Workflow_HITL/CreateApprovalRequest.cs @@ -0,0 +1,27 @@ +// Copyright (c) Microsoft. All rights reserved. + +using Microsoft.Agents.AI.Workflows; + +namespace SingleAgent; + +public record ApprovalRequest(string ExpenseId, decimal Amount, string EmployeeName); +public record ApprovalResponse(bool Approved, string? Comments); + +internal sealed class CreateApprovalRequest() : Executor("RetrieveRequest") +{ + public override async ValueTask HandleAsync(string message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + // Get request details from db. + return new ApprovalRequest(message, 1500.00m, "Jerry"); + } +} + +internal sealed class ExpenseReimburse() : Executor("Reimburse") +{ + public override async ValueTask HandleAsync(ApprovalResponse message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + // Simulate payment processing. + await Task.Delay(1000, cancellationToken); + return $"Expense reimbursed at {DateTime.Now.ToUniversalTime()}"; + } +} diff --git a/dotnet/samples/DurableAgents/ConsoleApps/10_Workflow_HITL/Program.cs b/dotnet/samples/DurableAgents/ConsoleApps/10_Workflow_HITL/Program.cs new file mode 100644 index 0000000000..c6d0a7d5fd --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/10_Workflow_HITL/Program.cs @@ -0,0 +1,105 @@ +// Copyright (c) Microsoft. All rights reserved. + +// This sample demonstrates a Human-in-the-Loop (HITL) workflow using Durable Tasks. +// The workflow creates an expense approval request, waits for manager approval via an external event, +// and then processes the expense reimbursement based on the approval response. +// This sample mirrors the pattern used in the in-process HumanInTheLoopBasic sample. + +using Microsoft.Agents.AI.DurableTask; +using Microsoft.Agents.AI.Workflows; +using Microsoft.DurableTask.Client; +using Microsoft.DurableTask.Client.AzureManaged; +using Microsoft.DurableTask.Worker.AzureManaged; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using SingleAgent; + +// Get DTS connection string from environment variable +string dtsConnectionString = Environment.GetEnvironmentVariable("DURABLE_TASK_SCHEDULER_CONNECTION_STRING") + ?? "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None"; + +// Define executors for the workflow +CreateApprovalRequest createRequest = new(); +RequestPort managerApproval = RequestPort.Create("ManagerApproval"); +ExpenseReimburse reimburse = new(); + +Workflow expenseApproval = new WorkflowBuilder(createRequest) + .WithName("ExpenseReImbursement") + .WithDescription("Expense ReImbursement") + .AddEdge(createRequest, managerApproval) + .AddEdge(managerApproval, reimburse) + .Build(); + +IHost host = Host.CreateDefaultBuilder(args) + .ConfigureLogging(logging => logging.SetMinimumLevel(LogLevel.Warning)) + .ConfigureServices(services => + { + services.ConfigureDurableWorkflows( + options => options.Workflows.AddWorkflow(expenseApproval), + workerBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString), + clientBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString)); + }) + .Build(); + +await host.StartAsync(); + +// Get services +DurableTaskClient durableClient = host.Services.GetRequiredService(); + +// Start the workflow with an expense ID as input +string expenseId = "EXP-2025-001"; +Console.WriteLine($"Starting expense reimbursement workflow for expense: {expenseId}"); + +// Start the workflow and get a streaming handle +await using DurableStreamingRun run = await DurableWorkflow.StreamAsync(expenseApproval, expenseId, durableClient); + +Console.WriteLine($"Workflow started with instance ID: {run.InstanceId}"); +Console.WriteLine("Watching for workflow events...\n"); + +// Watch for workflow events - similar pattern to InProcessExecution.StreamAsync +await foreach (WorkflowEvent evt in run.WatchStreamAsync()) +{ + switch (evt) + { + case DurableRequestInfoEvent requestEvent: + // Handle request for external input (human-in-the-loop) + Console.WriteLine($"Workflow is waiting for input at RequestPort: {requestEvent.RequestPortId}"); + Console.WriteLine($" Input data: {requestEvent.Input}"); + Console.WriteLine($" Expected response type: {requestEvent.ResponseType}"); + + // Simulate manager approval + ApprovalResponse response = HandleApprovalRequest(requestEvent); + await run.SendResponseAsync(requestEvent, response); + Console.WriteLine($" Response sent: Approved={response.Approved}\n"); + break; + + case DurableWorkflowCompletedEvent completedEvent: + // The workflow has completed + Console.WriteLine($"Workflow completed with result: {completedEvent.Result}"); + break; + + case DurableWorkflowFailedEvent failedEvent: + // The workflow has failed + Console.WriteLine($"Workflow failed: {failedEvent.ErrorMessage}"); + break; + } +} + +Console.ReadLine(); +await host.StopAsync(); + +// Handler for approval requests - similar to HandleExternalRequest in the in-process sample +static ApprovalResponse HandleApprovalRequest(DurableRequestInfoEvent requestEvent) +{ + // In a real scenario, this would involve human interaction (e.g., a web UI) + // For this sample, we simulate automatic approval + ApprovalRequest? request = requestEvent.GetInputAs(); + + if (request is not null) + { + Console.WriteLine($" Approval request for: {request.EmployeeName}, Amount: {request.Amount:C}"); + } + + return new ApprovalResponse(Approved: true, Comments: "Approved by manager. Looks good!"); +} diff --git a/dotnet/samples/DurableAgents/ConsoleApps/10_Workflow_HITL/README.md b/dotnet/samples/DurableAgents/ConsoleApps/10_Workflow_HITL/README.md new file mode 100644 index 0000000000..1c968b4be6 --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/10_Workflow_HITL/README.md @@ -0,0 +1,96 @@ +# Workflow Human-in-the-Loop (HITL) Sample + +This sample demonstrates a **Human-in-the-Loop** pattern in durable workflows using `RequestPort`. The workflow pauses execution to wait for external input (e.g., manager approval) and resumes when the response is provided. + +## Overview + +The sample implements an expense approval workflow: + +1. **CreateApprovalRequest** - Retrieves expense details and creates an approval request +2. **ManagerApproval** (RequestPort) - Pauses workflow to wait for manager approval +3. **ExpenseReimburse** - Processes the reimbursement based on approval response + +## Workflow Flow + +``` +User Input (Expense ID) + | + v ++---------------------+ +| CreateApprovalRequest| Creates ApprovalRequest with expense details ++---------------------+ + | + v ++---------------------+ +| ManagerApproval | RequestPort - PAUSES here waiting for external input +| (RequestPort) | Workflow is durable while waiting ++---------------------+ + | + v (ApprovalResponse) ++---------------------+ +| ExpenseReimburse | Processes reimbursement if approved ++---------------------+ + | + v + Result +``` + +## Key Concepts + +- **RequestPort** - A special executor that pauses the workflow and waits for external input +- **DurableRequestInfoEvent** - Event emitted when the workflow reaches a RequestPort +- **SendResponseAsync** - Method to provide the response and resume the workflow +- **Durability** - The workflow can survive process restarts while waiting for human input + +## Code Highlights + +### Defining the RequestPort + +```csharp +RequestPort managerApproval = + RequestPort.Create("ManagerApproval"); +``` + +### Handling the Request Event + +```csharp +await foreach (WorkflowEvent evt in run.WatchStreamAsync()) +{ + switch (evt) + { + case DurableRequestInfoEvent requestEvent: + // Workflow is waiting for input + ApprovalResponse response = HandleApprovalRequest(requestEvent); + await run.SendResponseAsync(requestEvent, response); + break; + // ... other events + } +} +``` + +## Environment Setup + +See the [README.md](../README.md) file in the parent directory for environment configuration. + +## Running the Sample + +```bash +cd dotnet/samples/DurableAgents/ConsoleApps/10_Workflow_HITL +dotnet run --framework net10.0 +``` + +### Sample Output + +``` +Starting expense reimbursement workflow for expense: EXP-2025-001 +Workflow started with instance ID: abc123... +Watching for workflow events... + +Workflow is waiting for input at RequestPort: ManagerApproval + Input data: {"ExpenseId":"EXP-2025-001","Amount":1500.00,"EmployeeName":"Jerry"} + Expected response type: SingleAgent.ApprovalResponse + Approval request for: Jerry, Amount: $1,500.00 + Response sent: Approved=True + +Workflow completed with result: Expense reimbursed at 1/23/2025 5:30:00 PM +``` diff --git a/dotnet/samples/DurableAgents/ConsoleApps/11_WorkflowEvents/11_WorkflowEvents.csproj b/dotnet/samples/DurableAgents/ConsoleApps/11_WorkflowEvents/11_WorkflowEvents.csproj new file mode 100644 index 0000000000..81439ef42b --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/11_WorkflowEvents/11_WorkflowEvents.csproj @@ -0,0 +1,29 @@ + + + net10.0 + Exe + enable + enable + SingleWorkflow + SingleAgent + + + + + + + + + + + + + + + + diff --git a/dotnet/samples/DurableAgents/ConsoleApps/11_WorkflowEvents/CancellationProgressEvent.cs b/dotnet/samples/DurableAgents/ConsoleApps/11_WorkflowEvents/CancellationProgressEvent.cs new file mode 100644 index 0000000000..1ce2f54359 --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/11_WorkflowEvents/CancellationProgressEvent.cs @@ -0,0 +1,16 @@ +// Copyright (c) Microsoft. All rights reserved. + +using Microsoft.Agents.AI.Workflows; + +namespace SingleAgent; + +/// +/// Event emitted to report cancellation progress. +/// +public sealed class CancellationProgressEvent(string orderId, int percentComplete, string status) + : WorkflowEvent($"Cancellation {percentComplete}%: {status}") +{ + public string OrderId { get; } = orderId; + public int PercentComplete { get; } = percentComplete; + public string Status { get; } = status; +} diff --git a/dotnet/samples/DurableAgents/ConsoleApps/11_WorkflowEvents/EmailSentEvent.cs b/dotnet/samples/DurableAgents/ConsoleApps/11_WorkflowEvents/EmailSentEvent.cs new file mode 100644 index 0000000000..24d209d985 --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/11_WorkflowEvents/EmailSentEvent.cs @@ -0,0 +1,14 @@ +// Copyright (c) Microsoft. All rights reserved. + +using Microsoft.Agents.AI.Workflows; + +namespace SingleAgent; + +/// +/// Event emitted when an email is sent. +/// +public sealed class EmailSentEvent(string email, string subject) : WorkflowEvent($"Email sent to {email}") +{ + public string Email { get; } = email; + public string Subject { get; } = subject; +} diff --git a/dotnet/samples/DurableAgents/ConsoleApps/11_WorkflowEvents/OrderCancelExecutor.cs b/dotnet/samples/DurableAgents/ConsoleApps/11_WorkflowEvents/OrderCancelExecutor.cs new file mode 100644 index 0000000000..b85541600b --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/11_WorkflowEvents/OrderCancelExecutor.cs @@ -0,0 +1,117 @@ +// Copyright (c) Microsoft. All rights reserved. + +// This sample demonstrates how to use IWorkflowContext methods in your executors: +// - AddEventAsync: Emit custom events that can be observed by the workflow caller +// - YieldOutputAsync: Stream intermediate outputs during execution +// +// These features enable rich observability and control over workflow execution. + +using Microsoft.Agents.AI.Workflows; + +namespace SingleAgent; + +#region Domain Models + +/// +/// Represents an order in the system. +/// +public sealed class Order +{ + public required string Id { get; set; } + public DateTime OrderDate { get; set; } + public bool IsCancelled { get; set; } + public required Customer Customer { get; set; } +} + +/// +/// Represents a customer associated with an order. +/// +public sealed class Customer +{ + public string Name { get; set; } = string.Empty; + public string Email { get; set; } = string.Empty; +} + +#endregion +#region Custom Workflow Events + +#endregion + +#region Executors + +/// +/// Looks up an order by its ID. Demonstrates AddEventAsync for custom events. +/// +internal sealed class OrderLookup() : Executor("OrderLookup") +{ + public override async ValueTask HandleAsync( + string message, + IWorkflowContext context, + CancellationToken cancellationToken = default) + { + await context.AddEventAsync(new OrderLookupStartedEvent(message), cancellationToken); + await Task.Delay(500, cancellationToken); + + Order order = new() + { + Id = message, + OrderDate = DateTime.UtcNow.AddDays(-3), + IsCancelled = false, + Customer = new Customer { Name = "Jerry", Email = "jerry@example.com" } + }; + + await context.AddEventAsync(new OrderFoundEvent(order), cancellationToken); + return order; + } +} + +/// +/// Cancels an order with progress reporting. +/// Demonstrates AddEventAsync for progress events and YieldOutputAsync for streaming outputs. +/// +internal sealed class OrderCancel() : Executor("OrderCancel") +{ + public override async ValueTask HandleAsync( + Order message, + IWorkflowContext context, + CancellationToken cancellationToken = default) + { + // Simulate cancellation steps with progress events + string[] steps = ["Validating", "Processing refund", "Finalizing"]; + for (int i = 0; i < steps.Length; i++) + { + await Task.Delay(500, cancellationToken); + int percent = (i + 1) * 33; + + // Emit progress event (callers can observe this in real-time) + await context.AddEventAsync(new CancellationProgressEvent(message.Id, percent, steps[i]), cancellationToken); + + // YieldOutputAsync streams intermediate results matching the executor's return type + await context.YieldOutputAsync(message, cancellationToken); + } + + message.IsCancelled = true; + await context.AddEventAsync(new OrderCancelledEvent(message.Id), cancellationToken); + return message; + } +} + +/// +/// Sends a cancellation confirmation email. Demonstrates AddEventAsync for completion events. +/// +internal sealed class SendEmail() : Executor("SendEmail") +{ + public override async ValueTask HandleAsync( + Order message, + IWorkflowContext context, + CancellationToken cancellationToken = default) + { + await Task.Delay(500, cancellationToken); + + string email = message.Customer.Email; + await context.AddEventAsync(new EmailSentEvent(email, $"Order {message.Id} Cancelled"), cancellationToken); + return $"Email sent to {email}"; + } +} + +#endregion diff --git a/dotnet/samples/DurableAgents/ConsoleApps/11_WorkflowEvents/OrderCancelledEvent.cs b/dotnet/samples/DurableAgents/ConsoleApps/11_WorkflowEvents/OrderCancelledEvent.cs new file mode 100644 index 0000000000..690209f4ed --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/11_WorkflowEvents/OrderCancelledEvent.cs @@ -0,0 +1,13 @@ +// Copyright (c) Microsoft. All rights reserved. + +using Microsoft.Agents.AI.Workflows; + +namespace SingleAgent; + +/// +/// Event emitted when an order is successfully cancelled. +/// +public sealed class OrderCancelledEvent(string orderId) : WorkflowEvent($"Order {orderId} has been cancelled") +{ + public string OrderId { get; } = orderId; +} diff --git a/dotnet/samples/DurableAgents/ConsoleApps/11_WorkflowEvents/OrderFoundEvent.cs b/dotnet/samples/DurableAgents/ConsoleApps/11_WorkflowEvents/OrderFoundEvent.cs new file mode 100644 index 0000000000..0d93bdf5d1 --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/11_WorkflowEvents/OrderFoundEvent.cs @@ -0,0 +1,13 @@ +// Copyright (c) Microsoft. All rights reserved. + +using Microsoft.Agents.AI.Workflows; + +namespace SingleAgent; + +/// +/// Event emitted when an order is found. +/// +public sealed class OrderFoundEvent(Order order) : WorkflowEvent($"Found order {order.Id} for {order.Customer.Name}") +{ + public Order Order { get; } = order; +} diff --git a/dotnet/samples/DurableAgents/ConsoleApps/11_WorkflowEvents/OrderLookupStartedEvent.cs b/dotnet/samples/DurableAgents/ConsoleApps/11_WorkflowEvents/OrderLookupStartedEvent.cs new file mode 100644 index 0000000000..3345165cef --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/11_WorkflowEvents/OrderLookupStartedEvent.cs @@ -0,0 +1,13 @@ +// Copyright (c) Microsoft. All rights reserved. + +using Microsoft.Agents.AI.Workflows; + +namespace SingleAgent; + +/// +/// Event emitted when an order lookup starts. +/// +public sealed class OrderLookupStartedEvent(string orderId) : WorkflowEvent($"Looking up order {orderId}") +{ + public string OrderId { get; } = orderId; +} diff --git a/dotnet/samples/DurableAgents/ConsoleApps/11_WorkflowEvents/Program.cs b/dotnet/samples/DurableAgents/ConsoleApps/11_WorkflowEvents/Program.cs new file mode 100644 index 0000000000..b96c77f053 --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/11_WorkflowEvents/Program.cs @@ -0,0 +1,135 @@ +// Copyright (c) Microsoft. All rights reserved. + +// ═══════════════════════════════════════════════════════════════════════════════ +// SAMPLE: Workflow Events and IWorkflowContext Features +// ═══════════════════════════════════════════════════════════════════════════════ +// +// This sample demonstrates how to use IWorkflowContext methods in executors: +// +// 1. AddEventAsync - Emit custom events that callers can observe in real-time +// 2. YieldOutputAsync - Stream intermediate outputs during long-running operations +// +// The sample uses DurableWorkflow.StreamAsync to observe events as they occur, +// showing how callers can receive real-time updates from the workflow. +// +// Workflow: OrderLookup -> OrderCancel -> SendEmail +// ═══════════════════════════════════════════════════════════════════════════════ + +using Microsoft.Agents.AI.DurableTask; +using Microsoft.Agents.AI.Workflows; +using Microsoft.DurableTask.Client; +using Microsoft.DurableTask.Client.AzureManaged; +using Microsoft.DurableTask.Worker.AzureManaged; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using SingleAgent; + +// Get DTS connection string from environment variable +string dtsConnectionString = Environment.GetEnvironmentVariable("DURABLE_TASK_SCHEDULER_CONNECTION_STRING") + ?? "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None"; + +// Define executors and build workflow +OrderLookup orderLookup = new(); +OrderCancel orderCancel = new(); +SendEmail sendEmail = new(); + +Workflow cancelOrder = new WorkflowBuilder(orderLookup) + .WithName("CancelOrder") + .WithDescription("Cancel an order and notify the customer") + .AddEdge(orderLookup, orderCancel) + .AddEdge(orderCancel, sendEmail) + .Build(); + +// Configure host with durable workflow support +IHost host = Host.CreateDefaultBuilder(args) + .ConfigureLogging(logging => logging.SetMinimumLevel(LogLevel.Warning)) + .ConfigureServices(services => + { + services.ConfigureDurableWorkflows( + options => options.Workflows.AddWorkflow(cancelOrder), + workerBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString), + clientBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString)); + }) + .Build(); + +await host.StartAsync(); + +DurableTaskClient durableClient = host.Services.GetRequiredService(); + +Console.WriteLine("Workflow Events Demo - Enter order ID (or 'exit'):"); + +while (true) +{ + Console.Write("> "); + string? input = Console.ReadLine(); + if (string.IsNullOrWhiteSpace(input) || input.Equals("exit", StringComparison.OrdinalIgnoreCase)) + { + break; + } + + try + { + await RunWorkflowWithStreamingAsync(input, cancelOrder, durableClient); + } + catch (Exception ex) + { + Console.WriteLine($"Error: {ex.Message}"); + } + + Console.WriteLine(); +} + +await host.StopAsync(); + +// Runs a workflow and streams events as they occur +async Task RunWorkflowWithStreamingAsync(string orderId, Workflow workflow, DurableTaskClient client) +{ + // StreamAsync starts the workflow and returns a handle for observing events + await using DurableStreamingRun run = await DurableWorkflow.StreamAsync(workflow, orderId, client); + Console.WriteLine($"Started: {run.InstanceId}"); + + // WatchStreamAsync yields events as they're emitted by executors + await foreach (WorkflowEvent evt in run.WatchStreamAsync()) + { + switch (evt) + { + // Custom domain events (emitted via AddEventAsync) + case OrderLookupStartedEvent e: + WriteColored($" [Lookup] Looking up order {e.OrderId}", ConsoleColor.Cyan); + break; + case OrderFoundEvent e: + WriteColored($" [Lookup] Found: {e.Order.Customer.Name}", ConsoleColor.Cyan); + break; + case CancellationProgressEvent e: + WriteColored($" [Cancel] {e.PercentComplete}% - {e.Status}", ConsoleColor.Yellow); + break; + case OrderCancelledEvent e: + WriteColored(" [Cancel] Done", ConsoleColor.Yellow); + break; + case EmailSentEvent e: + WriteColored($" [Email] Sent to {e.Email}", ConsoleColor.Magenta); + break; + + // Yielded outputs (emitted via YieldOutputAsync) + case DurableYieldedOutputEvent e: + WriteColored($" [Output] {e.ExecutorId}", ConsoleColor.DarkGray); + break; + + // Workflow completion + case DurableWorkflowCompletedEvent e: + WriteColored($"Completed: {e.Result}", ConsoleColor.Green); + break; + case DurableWorkflowFailedEvent e: + WriteColored($"Failed: {e.ErrorMessage}", ConsoleColor.Red); + break; + } + } +} + +void WriteColored(string message, ConsoleColor color) +{ + Console.ForegroundColor = color; + Console.WriteLine(message); + Console.ResetColor(); +} diff --git a/dotnet/samples/DurableAgents/ConsoleApps/11_WorkflowEvents/README.md b/dotnet/samples/DurableAgents/ConsoleApps/11_WorkflowEvents/README.md new file mode 100644 index 0000000000..0ea6210826 --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/11_WorkflowEvents/README.md @@ -0,0 +1,156 @@ +# Single Workflow Console Sample + +This sample demonstrates how to run a workflow as a durable orchestration from a console application using the Durable Task Framework. It showcases the **durability** aspect - if the process crashes mid-execution, the workflow can be resumed without re-executing completed activities. + +## Overview + +The sample implements an order cancellation workflow with three executors, each with artificial delays to simulate real-world operations: + +1. **OrderLookup** (2 seconds) - Looks up an order by its ID +2. **OrderCancel** (5 seconds) - Marks the order as cancelled +3. **SendEmail** (1 second) - Sends a cancellation confirmation email + +## Durability Demonstration + +The key feature of Durable Task Framework is **durability**: + + +- **Activity results are persisted**: When an activity completes, its result is saved +- **Orchestrations are replayed**: On restart, the orchestration replays from the beginning +- **Completed activities are skipped**: The framework uses cached results for completed activities +- **Failed activities are retried**: If an activity was interrupted, it runs again +- **Automatic resume**: When the worker starts, it automatically picks up any pending work! + +### Try It Yourself + +1. Start the application and enter an order ID (e.g., `12345`) +2. Stop the app (Ctrl+C or stop debugging) during the `OrderCancel` activity (5 seconds) +3. Restart the application +4. **Watch for automatic resume!** The worker automatically picks up the interrupted workflow +5. Observe that `OrderLookup` is NOT re-executed (its result was cached) +6. `OrderCancel` restarts from the beginning (it didn't complete) +7. `SendEmail` runs after `OrderCancel` completes + +The durability is completely automatic - no manual intervention needed! + +## Workflow Flow + +``` +User Input (Order ID) + ? + ? +??????????????????? +? OrderLookup ? ? 2 second delay (database lookup) +? (2 seconds) ? +??????????????????? + ? + ? +??????????????????? +? OrderCancel ? ? 5 second delay - TRY INTERRUPTING HERE! +? (5 seconds) ? +??????????????????? + ? + ? +??????????????????? +? SendEmail ? ? 1 second delay (email sending) +? (1 second) ? +??????????????????? + ? + ? + Result +``` + +## Key Concepts Demonstrated + +- **ConfigureDurableWorkflows** - Simplified API for registering workflows +- **DurableExecution.RunAsync** - Start a new workflow (similar to InProcessExecution) +- **DurableRun** - Handle to monitor and interact with a running workflow +- **Automatic Resume** - Interrupted workflows continue automatically on restart + +## 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, you can run the sample: + +```bash +cd dotnet/samples/DurableAgents/ConsoleApps/08_SingleWorkflow +dotnet run --framework net10.0 +``` + +### Sample Session + +```text +?????????????????????????????????????????????????????????????????????? +? Durable Workflow Console Sample ? +?????????????????????????????????????????????????????????????????????? +? This sample demonstrates durability in workflows. ? +? Workflow: OrderLookup (2s) -> OrderCancel (5s) -> SendEmail (1s) ? +?????????????????????????????????????????????????????????????????????? + +?? TIP: Stop the app during OrderCancel (5 seconds) to test durability! + Restart the app - it will automatically resume from where it left off. + +? Checking for pending workflows... + +Enter an order ID to start a new workflow (or 'exit' to quit): + +Order ID: 12345 + +Starting workflow for order '12345'... +Instance ID: abc123-def456-... + +??????????????????????????????????????????????????????????????????? +? [Activity] OrderLookup: Starting lookup for order '12345' +? [Activity] OrderLookup: Found order '12345' for customer 'Jerry' +??????????????????????????????????????????????????????????????????? + +??????????????????????????????????????????????????????????????????? +? [Activity] OrderCancel: Starting cancellation for order '12345' +? [Activity] OrderCancel: ?? This takes 5 seconds - try Ctrl+C! +? [Activity] OrderCancel: Processing... 1/5 seconds +? [Activity] OrderCancel: Processing... 2/5 seconds +^C <-- User stops the app here + +[After restart...] + +? Checking for pending workflows... + +??????????????????????????????????????????????????????????????????? +? [Activity] OrderCancel: Starting cancellation for order '12345' <-- Auto-resumed! +? [Activity] OrderCancel: ?? This takes 5 seconds - try Ctrl+C! +? [Activity] OrderCancel: Processing... 1/5 seconds +... +? [Activity] OrderCancel: ? Order '12345' has been cancelled +??????????????????????????????????????????????????????????????????? + +??????????????????????????????????????????????????????????????????? +? [Activity] SendEmail: Sending email to 'jerry@example.com'... +? [Activity] SendEmail: ? Email sent successfully! +??????????????????????????????????????????????????????????????????? + +Enter an order ID to start a new workflow (or 'exit' to quit): + +Order ID: _ +``` + +Notice that when resumed: +- `OrderLookup` was **NOT re-executed** (result was cached by Durable Task) +- `OrderCancel` **restarted automatically** (it was interrupted before completing) +- `SendEmail` ran normally after `OrderCancel` completed + +## Viewing Workflow State + +You can view the state of the workflow in the Durable Task Scheduler dashboard: + +1. Open your browser and navigate to `http://localhost:8082` +2. In the dashboard, you can view the state of the orchestration, including activity history and current state + +## Related Samples + +- [01_SingleAgent](../01_SingleAgent) - Single agent console sample +- [02_AgentOrchestration_Chaining](../02_AgentOrchestration_Chaining) - Agent chaining with durable orchestration +- [05_AgentOrchestration_HITL](../05_AgentOrchestration_HITL) - Human-in-the-loop orchestration +- [09_Workflow](../../AzureFunctions/09_Workflow) - Azure Functions version of workflow hosting diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableExecution.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableExecution.cs new file mode 100644 index 0000000000..e69de29bb2 diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableExecutorContext.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableExecutorContext.cs deleted file mode 100644 index 325145682e..0000000000 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableExecutorContext.cs +++ /dev/null @@ -1,269 +0,0 @@ -// 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.DurableTask; - -/// -/// 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. -/// -/// -/// 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.")] -public 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.DurableTask/DurableOptions.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableOptions.cs index 82bb8a419f..d737dfcf3e 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableOptions.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableOptions.cs @@ -20,7 +20,7 @@ public sealed class DurableOptions /// /// Initializes a new instance of the class. /// - internal DurableOptions() + public DurableOptions() { this.Workflows = new DurableWorkflowOptions(this); } diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableRun.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableRun.cs new file mode 100644 index 0000000000..5fe120e365 --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableRun.cs @@ -0,0 +1,283 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Diagnostics.CodeAnalysis; +using System.Text.Json; +using Microsoft.Agents.AI.Workflows; +using Microsoft.DurableTask.Client; + +namespace Microsoft.Agents.AI.DurableTask; + +/// +/// Represents a durable workflow run that tracks execution status and provides access to workflow events. +/// +/// +/// This class provides a similar API to but for workflows executed as durable orchestrations. +/// Events are received by raising external events to the orchestration and can be streamed to the caller. +/// +public sealed class DurableRun : IAsyncDisposable +{ + private readonly DurableTaskClient _client; + private readonly List _eventSink = []; + private int _lastBookmark; + + internal DurableRun(DurableTaskClient client, string instanceId, string workflowName) + { + this._client = client; + this.InstanceId = instanceId; + this.WorkflowName = workflowName; + } + + /// + /// Gets the unique instance ID for this orchestration run. + /// + public string InstanceId { get; } + + /// + /// Gets the name of the workflow being executed. + /// + public string WorkflowName { get; } + + /// + /// Gets the current execution status of the workflow run. + /// + /// A cancellation token to observe. + /// The current status of the durable run. + public async ValueTask GetStatusAsync(CancellationToken cancellationToken = default) + { + OrchestrationMetadata? metadata = await this._client.GetInstanceAsync( + this.InstanceId, + getInputsAndOutputs: false, + cancellation: cancellationToken).ConfigureAwait(false); + + if (metadata is null) + { + return DurableRunStatus.NotFound; + } + + return metadata.RuntimeStatus switch + { + OrchestrationRuntimeStatus.Pending => DurableRunStatus.Pending, + OrchestrationRuntimeStatus.Running => DurableRunStatus.Running, + OrchestrationRuntimeStatus.Completed => DurableRunStatus.Completed, + OrchestrationRuntimeStatus.Failed => DurableRunStatus.Failed, + OrchestrationRuntimeStatus.Terminated => DurableRunStatus.Terminated, + OrchestrationRuntimeStatus.Suspended => DurableRunStatus.Suspended, + _ => DurableRunStatus.Unknown + }; + } + + /// + /// Waits for the workflow to complete and returns the result. + /// + /// The expected result type. + /// A cancellation token to observe. + /// The result of the workflow execution. + /// Thrown when the workflow failed or was terminated. + public async ValueTask WaitForCompletionAsync(CancellationToken cancellationToken = default) + { + OrchestrationMetadata metadata = await this._client.WaitForInstanceCompletionAsync( + this.InstanceId, + getInputsAndOutputs: true, + cancellation: cancellationToken).ConfigureAwait(false); + + if (metadata.RuntimeStatus == OrchestrationRuntimeStatus.Completed) + { + return metadata.ReadOutputAs(); + } + + if (metadata.RuntimeStatus == OrchestrationRuntimeStatus.Failed) + { + string errorMessage = metadata.FailureDetails?.ErrorMessage ?? "Workflow execution failed."; + throw new InvalidOperationException(errorMessage); + } + + throw new InvalidOperationException($"Workflow ended with unexpected status: {metadata.RuntimeStatus}"); + } + + /// + /// Waits for the workflow to complete and returns the string result. + /// + /// A cancellation token to observe. + /// The string result of the workflow execution. + public ValueTask WaitForCompletionAsync(CancellationToken cancellationToken = default) + => this.WaitForCompletionAsync(cancellationToken); + + /// + /// Sends an external event to the workflow orchestration. + /// + /// + /// This can be used to send responses or messages to the workflow while it's running. + /// The orchestration must be waiting for the event using WaitForExternalEvent. + /// + /// The name of the event to raise. + /// The data to send with the event. + /// A cancellation token to observe. +#pragma warning disable CA1030 // Use events where appropriate - This is intentionally a method that sends events to an orchestration + public async ValueTask SendExternalEventAsync(string eventName, object? eventData = null, CancellationToken cancellationToken = default) +#pragma warning restore CA1030 + { + await this._client.RaiseEventAsync( + this.InstanceId, + eventName, + eventData, + cancellation: cancellationToken).ConfigureAwait(false); + } + + /// + /// Sends a workflow event to the orchestration. + /// + /// The workflow event to send. + /// A cancellation token to observe. + public ValueTask SendEventAsync(WorkflowEvent workflowEvent, CancellationToken cancellationToken = default) + => this.SendExternalEventAsync("WorkflowEvent", workflowEvent, cancellationToken); + + /// + /// Sends an external response to the workflow. + /// + /// The external response to send. + /// A cancellation token to observe. + public ValueTask SendResponseAsync(ExternalResponse response, CancellationToken cancellationToken = default) + => this.SendExternalEventAsync("ExternalResponse", response, cancellationToken); + + /// + /// Sends a response to a pending request port in the workflow (human-in-the-loop). + /// + /// + /// The response is serialized to JSON before being sent to match what the orchestration expects. + /// Use this method when responding to a that is waiting for external input. + /// + /// The type of the response data. + /// The ID of the request port to respond to. + /// The response data to send. + /// A cancellation token to observe. + [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Serializing workflow types provided by the caller.")] + [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Serializing workflow types provided by the caller.")] + public ValueTask SendResponseAsync(string requestPortId, TResponse response, CancellationToken cancellationToken = default) + { + ArgumentException.ThrowIfNullOrEmpty(requestPortId); + + // Serialize the response to JSON string - the orchestration expects a string via WaitForExternalEvent + string serializedResponse = JsonSerializer.Serialize(response); + return this.SendExternalEventAsync(requestPortId, serializedResponse, cancellationToken); + } + + /// + /// Gets all events that have been collected from the workflow. + /// + public IEnumerable OutgoingEvents => this._eventSink; + + /// + /// Gets the number of events collected since the last access to . + /// + public int NewEventCount => this._eventSink.Count - this._lastBookmark; + + /// + /// Gets all events collected since the last access to . + /// + public IEnumerable NewEvents + { + get + { + if (this._lastBookmark >= this._eventSink.Count) + { + return []; + } + + int currentBookmark = this._lastBookmark; + this._lastBookmark = this._eventSink.Count; + + return this._eventSink.Skip(currentBookmark); + } + } + + /// + /// Adds an event to the local event sink. + /// + /// + /// This is used internally to collect events raised by the orchestration. + /// In the durable scenario, events are typically returned as part of the orchestration output + /// or raised via external events. + /// + /// The event to add. + internal void AddEvent(WorkflowEvent workflowEvent) + { + this._eventSink.Add(workflowEvent); + } + + /// + /// Terminates the workflow orchestration. + /// + /// An optional reason for the termination. + /// A cancellation token to observe. + public async ValueTask TerminateAsync(string? reason = null, CancellationToken cancellationToken = default) + { + await this._client.TerminateInstanceAsync( + this.InstanceId, + reason, + cancellation: cancellationToken).ConfigureAwait(false); + } + + /// + /// Purges the orchestration instance history. + /// + /// A cancellation token to observe. + public async ValueTask PurgeAsync(CancellationToken cancellationToken = default) + { + await this._client.PurgeInstanceAsync( + this.InstanceId, + cancellation: cancellationToken).ConfigureAwait(false); + } + + /// + public ValueTask DisposeAsync() + { + // Nothing to dispose for durable runs - the orchestration continues independently + return default; + } +} + +/// +/// Represents the execution status of a durable workflow run. +/// +public enum DurableRunStatus +{ + /// + /// The orchestration instance was not found. + /// + NotFound, + + /// + /// The orchestration is pending and has not started. + /// + Pending, + + /// + /// The orchestration is currently running. + /// + Running, + + /// + /// The orchestration completed successfully. + /// + Completed, + + /// + /// The orchestration failed with an error. + /// + Failed, + + /// + /// The orchestration was terminated. + /// + Terminated, + + /// + /// The orchestration is suspended. + /// + Suspended, + + /// + /// The orchestration status is unknown. + /// + Unknown +} diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableStreamingRun.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableStreamingRun.cs new file mode 100644 index 0000000000..71f310210f --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableStreamingRun.cs @@ -0,0 +1,619 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Diagnostics.CodeAnalysis; +using System.Runtime.CompilerServices; +using System.Text.Json; +using Microsoft.Agents.AI.Workflows; +using Microsoft.DurableTask.Client; + +namespace Microsoft.Agents.AI.DurableTask; + +/// +/// Represents a durable workflow run that supports streaming workflow events as they occur. +/// +/// +/// This class provides a similar API to but for workflows executed as durable orchestrations. +/// Events are detected by monitoring the orchestration status for executors that are waiting +/// for external input (human-in-the-loop scenarios). +/// +public sealed class DurableStreamingRun : IAsyncDisposable +{ + private readonly DurableTaskClient _client; + private readonly Workflow _workflow; + private readonly List _requestPorts; + + internal DurableStreamingRun(DurableTaskClient client, string instanceId, Workflow workflow) + { + this._client = client; + this.InstanceId = instanceId; + this._workflow = workflow; + + // Extract RequestPorts from the workflow for event detection + this._requestPorts = ExtractRequestPorts(workflow); + } + + /// + /// Gets the unique instance ID for this orchestration run. + /// + public string InstanceId { get; } + + /// + /// Gets the name of the workflow being executed. + /// + public string WorkflowName => this._workflow.Name ?? string.Empty; + + /// + /// Gets the request ports defined in the workflow. + /// + public IReadOnlyList RequestPorts => this._requestPorts; + + /// + /// Gets the current execution status of the workflow run. + /// + /// A cancellation token to observe. + /// The current status of the durable run. + public async ValueTask GetStatusAsync(CancellationToken cancellationToken = default) + { + OrchestrationMetadata? metadata = await this._client.GetInstanceAsync( + this.InstanceId, + getInputsAndOutputs: false, + cancellation: cancellationToken).ConfigureAwait(false); + + if (metadata is null) + { + return DurableRunStatus.NotFound; + } + + return metadata.RuntimeStatus switch + { + OrchestrationRuntimeStatus.Pending => DurableRunStatus.Pending, + OrchestrationRuntimeStatus.Running => DurableRunStatus.Running, + OrchestrationRuntimeStatus.Completed => DurableRunStatus.Completed, + OrchestrationRuntimeStatus.Failed => DurableRunStatus.Failed, + OrchestrationRuntimeStatus.Terminated => DurableRunStatus.Terminated, + OrchestrationRuntimeStatus.Suspended => DurableRunStatus.Suspended, + _ => DurableRunStatus.Unknown + }; + } + + /// + /// Asynchronously streams workflow events as they occur during workflow execution. + /// + /// + /// + /// This method monitors the durable orchestration and yields instances + /// when the workflow reaches points that require external input (human-in-the-loop scenarios). + /// + /// + /// When the orchestration reaches a executor, a + /// is yielded containing the request data. The caller should then call + /// to provide the response and continue the workflow. + /// + /// + /// The interval between status checks. Defaults to 500ms. + /// A cancellation token to observe. + /// An asynchronous stream of objects. + public async IAsyncEnumerable WatchStreamAsync( + TimeSpan? pollingInterval = null, + [EnumeratorCancellation] CancellationToken cancellationToken = default) + { + TimeSpan interval = pollingInterval ?? TimeSpan.FromMilliseconds(500); + + // Track which request ports we've already yielded events for and are waiting for response + // Key: EventName (RequestPort ID), Value: Input data (to detect if we're at a different invocation) + Dictionary pendingRequests = []; + + // Track how many events we've already read from custom status + int lastReadEventIndex = 0; + + while (!cancellationToken.IsCancellationRequested) + { + OrchestrationMetadata? metadata = await this._client.GetInstanceAsync( + this.InstanceId, + getInputsAndOutputs: true, + cancellation: cancellationToken).ConfigureAwait(false); + + if (metadata is null) + { + yield break; + } + + // Check if the orchestration has completed + if (metadata.RuntimeStatus == OrchestrationRuntimeStatus.Completed) + { + yield return new DurableWorkflowCompletedEvent(metadata.SerializedOutput); + yield break; + } + + if (metadata.RuntimeStatus == OrchestrationRuntimeStatus.Failed) + { + string errorMessage = metadata.FailureDetails?.ErrorMessage ?? "Workflow execution failed."; + yield return new DurableWorkflowFailedEvent(errorMessage); + yield break; + } + + if (metadata.RuntimeStatus == OrchestrationRuntimeStatus.Terminated) + { + yield return new DurableWorkflowFailedEvent("Workflow was terminated."); + yield break; + } + + // Check custom status for events and pending external events + if (metadata.SerializedCustomStatus is not null) + { + DurableWorkflowCustomStatus? customStatus = TryParseCustomStatus(metadata.SerializedCustomStatus); + if (customStatus is not null) + { + // Yield any new events from executors + while (lastReadEventIndex < customStatus.Events.Count) + { + string serializedEvent = customStatus.Events[lastReadEventIndex]; + lastReadEventIndex++; + + WorkflowEvent? workflowEvent = TryDeserializeEvent(serializedEvent); + if (workflowEvent is not null) + { + yield return workflowEvent; + } + } + + // Check for pending external event (HITL) + if (customStatus.PendingEvent is not null) + { + PendingExternalEventStatus pendingStatus = customStatus.PendingEvent; + string eventName = pendingStatus.EventName; + string inputData = pendingStatus.Input; + + // Only yield a new event if: + // 1. We haven't seen this event name before, OR + // 2. The input data is different (meaning this is a new invocation of the same RequestPort) + bool shouldYield = !pendingRequests.TryGetValue(eventName, out string? previousInput) + || previousInput != inputData; + + if (shouldYield) + { + pendingRequests[eventName] = inputData; + + // Find the matching RequestPort + RequestPort? requestPort = this._requestPorts.Find(p => p.Id == eventName); + + yield return new DurableRequestInfoEvent( + RequestPortId: eventName, + Input: inputData, + RequestType: pendingStatus.RequestType, + ResponseType: pendingStatus.ResponseType, + RequestPort: requestPort); + } + } + } + else + { + // Try parsing as legacy PendingExternalEventStatus for backward compatibility + PendingExternalEventStatus? pendingStatus = TryParsePendingStatus(metadata.SerializedCustomStatus); + if (pendingStatus is not null) + { + string eventName = pendingStatus.EventName; + string inputData = pendingStatus.Input; + + bool shouldYield = !pendingRequests.TryGetValue(eventName, out string? previousInput) + || previousInput != inputData; + + if (shouldYield) + { + pendingRequests[eventName] = inputData; + RequestPort? requestPort = this._requestPorts.Find(p => p.Id == eventName); + + yield return new DurableRequestInfoEvent( + RequestPortId: eventName, + Input: inputData, + RequestType: pendingStatus.RequestType, + ResponseType: pendingStatus.ResponseType, + RequestPort: requestPort); + } + } + } + } + else + { + // Custom status is null - the orchestration is not waiting for external input + // Clear any pending requests that were waiting (they've been processed) + pendingRequests.Clear(); + } + + await Task.Delay(interval, cancellationToken).ConfigureAwait(false); + } + } + + [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow custom status.")] + [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing workflow custom status.")] + private static DurableWorkflowCustomStatus? TryParseCustomStatus(string serializedStatus) + { + try + { + return JsonSerializer.Deserialize(serializedStatus); + } + catch (JsonException) + { + return null; + } + } + + [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow event types.")] + [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing workflow event types.")] + [UnconditionalSuppressMessage("Trimming", "IL2057", Justification = "Event types are registered at startup and available at runtime.")] + private static WorkflowEvent? TryDeserializeEvent(string serializedEvent) + { + try + { + // First try to deserialize as SerializedWorkflowEvent (new format with type info) + DurableWorkflowRunner.SerializedWorkflowEvent? wrapper = + JsonSerializer.Deserialize(serializedEvent); + + if (wrapper?.TypeName is not null && wrapper.Data is not null) + { + Type? eventType = Type.GetType(wrapper.TypeName); + if (eventType is not null) + { + // Use custom deserialization for event types with constructor parameter mismatches + return DeserializeEventByType(eventType, wrapper.Data); + } + } + + // Fall back to deserializing as base WorkflowEvent (legacy format) + return JsonSerializer.Deserialize(serializedEvent); + } + catch (JsonException) + { + return null; + } + } + + /// + /// Deserializes an event by type, handling constructor parameter name mismatches. + /// + [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow event types.")] + [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing workflow event types.")] + private static WorkflowEvent? DeserializeEventByType(Type eventType, string json) + { + using JsonDocument doc = JsonDocument.Parse(json); + JsonElement root = doc.RootElement; + + // Handle ExecutorInvokedEvent: constructor expects (executorId, message) but JSON has (ExecutorId, Data) + if (eventType == typeof(ExecutorInvokedEvent)) + { + string executorId = root.GetProperty("ExecutorId").GetString() ?? string.Empty; + JsonElement? data = GetDataProperty(root); + return new ExecutorInvokedEvent(executorId, data!); + } + + // Handle ExecutorCompletedEvent: constructor expects (executorId, result) but JSON has (ExecutorId, Data) + if (eventType == typeof(ExecutorCompletedEvent)) + { + string executorId = root.GetProperty("ExecutorId").GetString() ?? string.Empty; + JsonElement? data = GetDataProperty(root); + return new ExecutorCompletedEvent(executorId, data); + } + + // For other event types, try standard deserialization with case-insensitive options + return JsonSerializer.Deserialize(json, eventType, s_caseInsensitiveOptions) as WorkflowEvent; + } + + // Cached JsonSerializerOptions for case-insensitive deserialization + private static readonly JsonSerializerOptions s_caseInsensitiveOptions = new() { PropertyNameCaseInsensitive = true }; + + /// + /// Gets the Data property from a JSON element. + /// + private static JsonElement? GetDataProperty(JsonElement root) + { + if (!root.TryGetProperty("Data", out JsonElement dataElement)) + { + return null; + } + + if (dataElement.ValueKind == JsonValueKind.Null) + { + return null; + } + + return dataElement.Clone(); + } + + [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing known type PendingExternalEventStatus.")] + [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing known type PendingExternalEventStatus.")] + private static PendingExternalEventStatus? TryParsePendingStatus(string serializedStatus) + { + try + { + return JsonSerializer.Deserialize(serializedStatus); + } + catch (JsonException) + { + return null; + } + } + + /// + /// Sends an external event to the workflow orchestration. + /// + /// The name of the event to raise (typically the ). + /// The data to send with the event. + /// A cancellation token to observe. +#pragma warning disable CA1030 // Use events where appropriate + public async ValueTask SendExternalEventAsync(string eventName, object? eventData = null, CancellationToken cancellationToken = default) +#pragma warning restore CA1030 + { + await this._client.RaiseEventAsync( + this.InstanceId, + eventName, + eventData, + cancellation: cancellationToken).ConfigureAwait(false); + } + + /// + /// Sends a response to a pending request in the workflow. + /// + /// + /// The response is serialized to JSON before being sent to match what the orchestration expects. + /// + /// The type of the response data. + /// The ID of the request port to respond to. + /// The response data to send. + /// A cancellation token to observe. + [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Serializing workflow types provided by the caller.")] + [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Serializing workflow types provided by the caller.")] + public ValueTask SendResponseAsync(string requestPortId, TResponse response, CancellationToken cancellationToken = default) + { + ArgumentException.ThrowIfNullOrEmpty(requestPortId); + + // Serialize the response to JSON string - the orchestration expects a string via WaitForExternalEvent + string serializedResponse = JsonSerializer.Serialize(response); + return this.SendExternalEventAsync(requestPortId, serializedResponse, cancellationToken); + } + + /// + /// Sends a response to a . + /// + /// + /// The response is serialized to JSON before being sent to match what the orchestration expects. + /// + /// The type of the response data. + /// The request event to respond to. + /// The response data to send. + /// A cancellation token to observe. + [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Serializing workflow types provided by the caller.")] + [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Serializing workflow types provided by the caller.")] + public ValueTask SendResponseAsync(DurableRequestInfoEvent requestEvent, TResponse response, CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(requestEvent); + + // Serialize the response to JSON string - the orchestration expects a string via WaitForExternalEvent + string serializedResponse = JsonSerializer.Serialize(response); + return this.SendExternalEventAsync(requestEvent.RequestPortId, serializedResponse, cancellationToken); + } + + /// + /// Waits for the workflow to complete and returns the result. + /// + /// The expected result type. + /// A cancellation token to observe. + /// The result of the workflow execution. + public async ValueTask WaitForCompletionAsync(CancellationToken cancellationToken = default) + { + OrchestrationMetadata metadata = await this._client.WaitForInstanceCompletionAsync( + this.InstanceId, + getInputsAndOutputs: true, + cancellation: cancellationToken).ConfigureAwait(false); + + if (metadata.RuntimeStatus == OrchestrationRuntimeStatus.Completed) + { + return metadata.ReadOutputAs(); + } + + if (metadata.RuntimeStatus == OrchestrationRuntimeStatus.Failed) + { + string errorMessage = metadata.FailureDetails?.ErrorMessage ?? "Workflow execution failed."; + throw new InvalidOperationException(errorMessage); + } + + throw new InvalidOperationException($"Workflow ended with unexpected status: {metadata.RuntimeStatus}"); + } + + /// + /// Waits for the workflow to complete and returns the string result. + /// + /// A cancellation token to observe. + /// The string result of the workflow execution. + public ValueTask WaitForCompletionAsync(CancellationToken cancellationToken = default) + => this.WaitForCompletionAsync(cancellationToken); + + /// + /// Terminates the workflow orchestration. + /// + /// An optional reason for the termination. + /// A cancellation token to observe. + public async ValueTask TerminateAsync(string? reason = null, CancellationToken cancellationToken = default) + { + await this._client.TerminateInstanceAsync( + this.InstanceId, + reason, + cancellation: cancellationToken).ConfigureAwait(false); + } + + /// + public ValueTask DisposeAsync() + { + // Nothing to dispose for durable runs - the orchestration continues independently + return default; + } + + private static List ExtractRequestPorts(Workflow workflow) + { + List requestPorts = []; + + foreach (WorkflowExecutorInfo executorInfo in WorkflowHelper.GetExecutorsFromWorkflowInOrder(workflow)) + { + if (executorInfo.RequestPort is not null) + { + requestPorts.Add(executorInfo.RequestPort); + } + } + + return requestPorts; + } +} + +/// +/// Event raised when the durable workflow is waiting for external input at a . +/// +/// The ID of the request port waiting for input. +/// The serialized input data that was passed to the RequestPort. +/// The full type name of the request type. +/// The full type name of the expected response type. +/// The request port definition, if available. +public sealed class DurableRequestInfoEvent( + string RequestPortId, + string Input, + string RequestType, + string ResponseType, + RequestPort? RequestPort) : WorkflowEvent(Input) +{ + /// + /// Gets the ID of the request port waiting for input. + /// + public string RequestPortId { get; } = RequestPortId; + + /// + /// Gets the serialized input data that was passed to the RequestPort. + /// + public string Input { get; } = Input; + + /// + /// Gets the full type name of the request type. + /// + public string RequestType { get; } = RequestType; + + /// + /// Gets the full type name of the expected response type. + /// + public string ResponseType { get; } = ResponseType; + + /// + /// Gets the request port definition, if available. + /// + public RequestPort? RequestPort { get; } = RequestPort; + + /// + /// Attempts to deserialize the input data to the specified type. + /// + /// The type to deserialize to. + /// The deserialized input, or default if deserialization fails. + [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow types provided by the caller.")] + [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing workflow types provided by the caller.")] + public T? GetInputAs() + { + try + { + return JsonSerializer.Deserialize(this.Input); + } + catch (JsonException) + { + return default; + } + } +} + +/// +/// Event raised when a durable workflow completes successfully. +/// +public sealed class DurableWorkflowCompletedEvent : WorkflowEvent +{ + /// + /// Initializes a new instance of the class. + /// + /// The serialized result of the workflow. + public DurableWorkflowCompletedEvent(string? result) : base(result) + { + this.Result = result; + } + + /// + /// Gets the serialized result of the workflow. + /// + public string? Result { get; } +} + +/// +/// Event raised when a durable workflow fails. +/// +public sealed class DurableWorkflowFailedEvent : WorkflowEvent +{ + /// + /// Initializes a new instance of the class. + /// + /// The error message describing the failure. + public DurableWorkflowFailedEvent(string errorMessage) : base(errorMessage) + { + this.ErrorMessage = errorMessage; + } + + /// + /// Gets the error message describing the failure. + /// + public string ErrorMessage { get; } +} + +/// +/// Event raised when an executor yields intermediate output via . +/// +/// +/// This is the durable equivalent of since that class has an internal +/// constructor not accessible from outside the Workflows assembly. +/// +public sealed class DurableYieldedOutputEvent : WorkflowEvent +{ + /// + /// Initializes a new instance of the class. + /// + /// The ID of the executor that yielded the output. + /// The yielded output value. + public DurableYieldedOutputEvent(string executorId, object output) : base(output) + { + this.ExecutorId = executorId; + this.Output = output; + } + + /// + /// Gets the ID of the executor that yielded the output. + /// + public string ExecutorId { get; } + + /// + /// Gets the yielded output value. + /// + public object Output { get; } +} + +/// +/// Event raised when an executor requests the workflow to halt via . +/// +/// +/// This is the durable equivalent of the internal RequestHaltEvent since that class is not accessible +/// from outside the Workflows assembly. +/// +public sealed class DurableHaltRequestedEvent : WorkflowEvent +{ + /// + /// Initializes a new instance of the class. + /// + /// The ID of the executor that requested the halt. + public DurableHaltRequestedEvent(string executorId) : base($"Halt requested by {executorId}") + { + this.ExecutorId = executorId; + } + + /// + /// Gets the ID of the executor that requested the halt. + /// + public string ExecutorId { get; } +} diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflow.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflow.cs new file mode 100644 index 0000000000..cac18319d9 --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflow.cs @@ -0,0 +1,158 @@ +// Copyright (c) Microsoft. All rights reserved. + +using Microsoft.Agents.AI.Workflows; +using Microsoft.DurableTask.Client; + +namespace Microsoft.Agents.AI.DurableTask; + +/// +/// Provides methods to run workflows as durable orchestrations. +/// +public static class DurableWorkflow +{ + /// + /// Runs a workflow as a durable orchestration and returns a handle to monitor its execution. + /// + /// The type of the input to the workflow. + /// The workflow to execute. + /// The input to pass to the workflow's starting executor. + /// The durable task client for orchestration operations. + /// Optional instance ID for the orchestration. If not provided, a new ID will be generated. + /// A cancellation token to observe. + /// A that can be used to monitor the workflow execution. + /// Thrown when workflow or client is null. + /// Thrown when the workflow does not have a valid name. + public static async ValueTask RunAsync( + Workflow workflow, + TInput input, + DurableTaskClient client, + string? instanceId = null, + CancellationToken cancellationToken = default) + where TInput : notnull + { + ArgumentNullException.ThrowIfNull(workflow); + ArgumentNullException.ThrowIfNull(client); + + if (string.IsNullOrEmpty(workflow.Name)) + { + throw new ArgumentException("Workflow must have a valid Name property.", nameof(workflow)); + } + + string orchestrationName = WorkflowNamingHelper.ToOrchestrationFunctionName(workflow.Name); + string actualInstanceId = await client.ScheduleNewOrchestrationInstanceAsync( + orchestratorName: orchestrationName, + input: input, + cancellation: cancellationToken).ConfigureAwait(false); + + return new DurableRun(client, actualInstanceId, workflow.Name); + } + + /// + /// Runs a workflow as a durable orchestration with string input. + /// + /// The workflow to execute. + /// The string input to pass to the workflow. + /// The durable task client for orchestration operations. + /// Optional instance ID for the orchestration. + /// A cancellation token to observe. + /// A that can be used to monitor the workflow execution. + public static ValueTask RunAsync( + Workflow workflow, + string input, + DurableTaskClient client, + string? instanceId = null, + CancellationToken cancellationToken = default) + => RunAsync(workflow, input, client, instanceId, cancellationToken); + + /// + /// Starts a workflow as a durable orchestration and returns a streaming handle to watch events. + /// + /// The type of the input to the workflow. + /// The workflow to execute. + /// The input to pass to the workflow's starting executor. + /// The durable task client for orchestration operations. + /// Optional instance ID for the orchestration. If not provided, a new ID will be generated. + /// A cancellation token to observe. + /// A that can be used to stream workflow events. + /// Thrown when workflow or client is null. + /// Thrown when the workflow does not have a valid name. + public static async ValueTask StreamAsync( + Workflow workflow, + TInput input, + DurableTaskClient client, + string? instanceId = null, + CancellationToken cancellationToken = default) + where TInput : notnull + { + ArgumentNullException.ThrowIfNull(workflow); + ArgumentNullException.ThrowIfNull(client); + + if (string.IsNullOrEmpty(workflow.Name)) + { + throw new ArgumentException("Workflow must have a valid Name property.", nameof(workflow)); + } + + string orchestrationName = WorkflowNamingHelper.ToOrchestrationFunctionName(workflow.Name); + string actualInstanceId = await client.ScheduleNewOrchestrationInstanceAsync( + orchestratorName: orchestrationName, + input: input, + cancellation: cancellationToken).ConfigureAwait(false); + + return new DurableStreamingRun(client, actualInstanceId, workflow); + } + + /// + /// Starts a workflow as a durable orchestration with string input and returns a streaming handle. + /// + /// The workflow to execute. + /// The string input to pass to the workflow. + /// The durable task client for orchestration operations. + /// Optional instance ID for the orchestration. + /// A cancellation token to observe. + /// A that can be used to stream workflow events. + public static ValueTask StreamAsync( + Workflow workflow, + string input, + DurableTaskClient client, + string? instanceId = null, + CancellationToken cancellationToken = default) + => StreamAsync(workflow, input, client, instanceId, cancellationToken); + + /// + /// Attaches to an existing workflow orchestration instance. + /// + /// The instance ID of the orchestration to attach to. + /// The name of the workflow being executed. + /// The durable task client for orchestration operations. + /// A that can be used to monitor the workflow execution. + public static DurableRun Attach( + string instanceId, + string workflowName, + DurableTaskClient client) + { + ArgumentException.ThrowIfNullOrEmpty(instanceId); + ArgumentException.ThrowIfNullOrEmpty(workflowName); + ArgumentNullException.ThrowIfNull(client); + + return new DurableRun(client, instanceId, workflowName); + } + + /// + /// Attaches to an existing workflow orchestration instance for streaming. + /// + /// The instance ID of the orchestration to attach to. + /// The workflow being executed. + /// The durable task client for orchestration operations. + /// A that can be used to stream workflow events. + public static DurableStreamingRun AttachStream( + string instanceId, + Workflow workflow, + DurableTaskClient client) + { + ArgumentException.ThrowIfNullOrEmpty(instanceId); + ArgumentNullException.ThrowIfNull(workflow); + ArgumentNullException.ThrowIfNull(client); + + return new DurableStreamingRun(client, instanceId, workflow); + } +} diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunner.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunner.cs index a5284fc508..4e1cbd559d 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunner.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunner.cs @@ -4,11 +4,39 @@ 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; +/// +/// Represents the custom status set when the orchestration is waiting for an external event. +/// +/// The name of the event being waited for (the RequestPort ID). +/// The serialized input data that was passed to the RequestPort. +/// The full type name of the request type. +/// The full type name of the expected response type. +public sealed record PendingExternalEventStatus( + string EventName, + string Input, + string RequestType, + string ResponseType); + +/// +/// Represents the complete custom status for a durable workflow orchestration. +/// +public sealed class DurableWorkflowCustomStatus +{ + /// + /// Gets or sets the pending external event status when waiting for HITL input. + /// + public PendingExternalEventStatus? PendingEvent { get; set; } + + /// + /// Gets or sets the list of serialized workflow events emitted by executors. + /// + public List Events { get; set; } = []; +} + /// /// Core workflow runner that executes workflow orchestrations using Durable Tasks. /// This class contains the core workflow execution logic independent of the hosting environment. @@ -64,23 +92,7 @@ public class DurableWorkflowRunner logger.LogRunningWorkflow(workflow.Name); - string result = await this.ExecuteWorkflowLevelsAsync(context, workflow, 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); + return await this.ExecuteWorkflowLevelsAsync(context, workflow, input, logger).ConfigureAwait(true); } /// @@ -181,6 +193,10 @@ public class DurableWorkflowRunner WorkflowExecutionPlan plan = WorkflowHelper.GetExecutionPlan(workflow); Dictionary results = new(plan.Levels.Sum(l => l.Executors.Count)); + // Track accumulated events and shared state + DurableWorkflowCustomStatus customStatus = new(); + Dictionary sharedState = []; + foreach (WorkflowExecutionLevel level in plan.Levels) { // Filter executors based on edge conditions from their predecessors @@ -196,29 +212,173 @@ public class DurableWorkflowRunner { WorkflowExecutorInfo executorInfo = eligibleExecutors[0]; string input = GetExecutorInput(executorInfo.ExecutorId, initialInput, results, plan); - results[executorInfo.ExecutorId] = await this.ExecuteExecutorAsync(context, executorInfo, input, logger).ConfigureAwait(true); + string rawResult = await this.ExecuteExecutorAsync(context, executorInfo, input, logger, customStatus, sharedState).ConfigureAwait(true); + results[executorInfo.ExecutorId] = UnwrapActivityResult(rawResult, customStatus, sharedState); + + // Update custom status with any new events + UpdateCustomStatus(context, customStatus); } else { + // For parallel execution, each activity gets a snapshot of the current state + // State updates are merged after all activities complete Task<(string Id, string Result)>[] tasks = new Task<(string Id, string Result)>[eligibleExecutors.Count]; for (int i = 0; i < eligibleExecutors.Count; i++) { WorkflowExecutorInfo executorInfo = eligibleExecutors[i]; string input = GetExecutorInput(executorInfo.ExecutorId, initialInput, results, plan); - tasks[i] = this.ExecuteExecutorWithIdAsync(context, executorInfo, input, logger); + tasks[i] = this.ExecuteExecutorWithIdAsync(context, executorInfo, input, logger, customStatus, sharedState); } (string Id, string Result)[] completedTasks = await Task.WhenAll(tasks).ConfigureAwait(true); - foreach ((string id, string result) in completedTasks) + foreach ((string id, string rawResult) in completedTasks) { - results[id] = result; + results[id] = UnwrapActivityResult(rawResult, customStatus, sharedState); } + + // Update custom status with any new events + UpdateCustomStatus(context, customStatus); } } return GetFinalResult(plan, results); } + /// + /// Unwraps an activity result, extracting state updates, events, and returning the actual result. + /// + [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing known wrapper type.")] + [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing known wrapper type.")] + private static string UnwrapActivityResult(string rawResult, DurableWorkflowCustomStatus customStatus, Dictionary sharedState) + { + if (string.IsNullOrEmpty(rawResult)) + { + return rawResult; + } + + try + { + // Try to deserialize as DurableActivityOutput + DurableActivityOutput? output = JsonSerializer.Deserialize(rawResult); + + // Check if this is actually a DurableActivityOutput (has Result property set or state updates) + // This distinguishes it from other JSON objects that would deserialize with default/empty values + if (output is not null && (output.Result is not null || output.StateUpdates.Count > 0 || output.ClearedScopes.Count > 0 || output.Events.Count > 0)) + { + // Apply cleared scopes first + foreach (string clearedScope in output.ClearedScopes) + { + string scopePrefix = clearedScope == "__default__" ? "__default__:" : $"{clearedScope}:"; + List keysToRemove = sharedState.Keys + .Where(k => k.StartsWith(scopePrefix, StringComparison.Ordinal)) + .ToList(); + + foreach (string key in keysToRemove) + { + sharedState.Remove(key); + } + } + + // Apply state updates + foreach (KeyValuePair update in output.StateUpdates) + { + if (update.Value is null) + { + sharedState.Remove(update.Key); + } + else + { + sharedState[update.Key] = update.Value; + } + } + + // Add events to the accumulated list + if (output.Events.Count > 0) + { + customStatus.Events.AddRange(output.Events); + } + + return output.Result ?? string.Empty; + } + } + catch (JsonException) + { + // Not a wrapped result, return as-is + } + + return rawResult; + } + + /// + /// Updates the orchestration custom status with current events and pending event info. + /// + private static void UpdateCustomStatus(TaskOrchestrationContext context, DurableWorkflowCustomStatus customStatus) + { + // Only update if there are events or a pending event + if (customStatus.Events.Count > 0 || customStatus.PendingEvent is not null) + { + context.SetCustomStatus(customStatus); + } + } + + /// + /// Wrapper for activity output that includes state updates and events. + /// + internal sealed class ActivityOutputWithState + { + /// + /// Gets or sets the serialized result of the activity. + /// + public string? Result { get; set; } + + /// + /// Gets or sets state updates made during activity execution. + /// + public Dictionary StateUpdates { get; set; } = []; + + /// + /// Gets or sets scopes that were cleared during activity execution. + /// + public List ClearedScopes { get; set; } = []; + + /// + /// Gets or sets the serialized workflow events emitted during activity execution. + /// + public List Events { get; set; } = []; + } + + /// + /// Wrapper for serialized workflow events that includes type information for proper deserialization. + /// + public sealed class SerializedWorkflowEvent + { + /// + /// Gets or sets the assembly-qualified type name of the event. + /// + public string? TypeName { get; set; } + + /// + /// Gets or sets the serialized JSON data of the event. + /// + public string? Data { get; set; } + } + + /// + /// Wrapper for activity input that includes shared state. + /// + internal sealed class ActivityInputWithState + { + /// + /// Gets or sets the serialized executor input. + /// + public string? Input { get; set; } + + /// + /// Gets or sets the shared state dictionary. + /// + public Dictionary State { get; set; } = []; + } + /// /// Filters executors based on their incoming edge conditions. /// An executor is eligible if all its incoming edges have conditions that evaluate to true, @@ -323,28 +483,84 @@ public class DurableWorkflowRunner TaskOrchestrationContext context, WorkflowExecutorInfo executorInfo, string input, - ILogger logger) + ILogger logger, + DurableWorkflowCustomStatus customStatus, + Dictionary sharedState) { - string result = await this.ExecuteExecutorAsync(context, executorInfo, input, logger).ConfigureAwait(true); + string result = await this.ExecuteExecutorAsync(context, executorInfo, input, logger, customStatus, sharedState).ConfigureAwait(true); return (executorInfo.ExecutorId, result); } + [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Serializing known wrapper type.")] + [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Serializing known wrapper type.")] private async Task ExecuteExecutorAsync( TaskOrchestrationContext context, WorkflowExecutorInfo executorInfo, string input, - ILogger logger) + ILogger logger, + DurableWorkflowCustomStatus customStatus, + Dictionary sharedState) { + // Handle RequestPort executors by waiting for external event (human-in-the-loop) + if (executorInfo.IsRequestPortExecutor) + { + return await ExecuteRequestPortAsync(context, executorInfo, input, logger, customStatus).ConfigureAwait(true); + } + if (!executorInfo.IsAgenticExecutor) { string executorName = WorkflowNamingHelper.GetExecutorName(executorInfo.ExecutorId); string triggerName = WorkflowNamingHelper.ToOrchestrationFunctionName(executorName); - return await context.CallActivityAsync(triggerName, input).ConfigureAwait(true); + + // Wrap input with shared state for the activity + ActivityInputWithState inputWithState = new() + { + Input = input, + State = new Dictionary(sharedState) // Pass a copy of the state + }; + + string wrappedInput = JsonSerializer.Serialize(inputWithState); + return await context.CallActivityAsync(triggerName, wrappedInput).ConfigureAwait(true); } return await ExecuteAgentAsync(context, executorInfo, input, logger).ConfigureAwait(true); } + private static async Task ExecuteRequestPortAsync( + TaskOrchestrationContext context, + WorkflowExecutorInfo executorInfo, + string input, + ILogger logger, + DurableWorkflowCustomStatus customStatus) + { + RequestPort requestPort = executorInfo.RequestPort!; + string eventName = requestPort.Id; + + logger.LogWaitingForExternalEvent(eventName, input); + + // Set custom status to notify clients that we're waiting for external input + // Include any accumulated events + customStatus.PendingEvent = new( + EventName: eventName, + Input: input, + RequestType: requestPort.Request.FullName ?? requestPort.Request.Name, + ResponseType: requestPort.Response.FullName ?? requestPort.Response.Name); + + context.SetCustomStatus(customStatus); + + // Wait for the external event (human-in-the-loop) + // The event data will be the response from the external actor + string response = await context.WaitForExternalEvent(eventName).ConfigureAwait(true); + + // Clear pending event status after receiving the event + customStatus.PendingEvent = null; + context.SetCustomStatus(customStatus.Events.Count > 0 ? customStatus : null); + + logger.LogReceivedExternalEvent(eventName, response); + + return response; + } + private static async Task ExecuteAgentAsync( TaskOrchestrationContext context, WorkflowExecutorInfo executorInfo, @@ -360,8 +576,8 @@ public class DurableWorkflowRunner return $"Agent '{agentName}' not found"; } - AgentThread thread = agent.GetNewThread(); - AgentRunResponse response = await agent.RunAsync(input, thread).ConfigureAwait(true); + AgentThread thread = await agent.GetNewThreadAsync(); + AgentResponse response = await agent.RunAsync(input, thread); return response.Text; } diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowServiceCollectionExtensions.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowServiceCollectionExtensions.cs new file mode 100644 index 0000000000..aa358affb9 --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowServiceCollectionExtensions.cs @@ -0,0 +1,389 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Diagnostics.CodeAnalysis; +using System.Text.Json; +using System.Text.Json.Serialization.Metadata; +using Microsoft.Agents.AI.DurableTask.State; +using Microsoft.Agents.AI.Workflows; +using Microsoft.Agents.AI.Workflows.Checkpointing; +using Microsoft.DurableTask; +using Microsoft.DurableTask.Client; +using Microsoft.DurableTask.Worker; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; + +namespace Microsoft.Agents.AI.DurableTask; + +/// +/// Extension methods for configuring durable workflows with the service collection. +/// +public static class DurableWorkflowServiceCollectionExtensions +{ + /// + /// Configures durable workflows with the service collection, automatically registering + /// orchestrations and activities for each workflow. + /// + /// The service collection to configure. + /// A delegate to configure the durable options. + /// An optional delegate to configure the durable task worker. + /// An optional delegate to configure the durable task client. + /// The service collection for chaining. + public static IServiceCollection ConfigureDurableWorkflows( + this IServiceCollection services, + Action configure, + Action? workerBuilder = null, + Action? clientBuilder = null) + { + ArgumentNullException.ThrowIfNull(services); + ArgumentNullException.ThrowIfNull(configure); + + // Create and configure durable options + DurableOptions durableOptions = new(); + configure(durableOptions); + + // Register DurableOptions as a singleton + services.AddSingleton(durableOptions); + + // Register the workflow runner + services.AddSingleton(); + + // Build registration info for all workflows + List registrations = []; + HashSet registeredActivities = []; + + foreach (KeyValuePair workflowEntry in durableOptions.Workflows.Workflows) + { + registrations.Add(BuildWorkflowRegistration(workflowEntry.Value, registeredActivities)); + } + + // Get any AI agents that were auto-registered from workflows + IReadOnlyDictionary> agentFactories = durableOptions.Agents.GetAgentFactories(); + + // Configure Durable Task Worker with orchestrations and activities + services.AddDurableTaskWorker(builder => + { + workerBuilder?.Invoke(builder); + + builder.AddTasks(registry => + { + // Register all workflow tasks + foreach (WorkflowRegistrationInfo registration in registrations) + { + // Register orchestration + registry.AddOrchestratorFunc( + registration.OrchestrationName, + (context, input) => RunWorkflowOrchestrationAsync(context, input, durableOptions)); + + // Register activities + foreach (ActivityRegistrationInfo activity in registration.Activities) + { + ExecutorBinding binding = activity.Binding; + registry.AddActivityFunc( + activity.ActivityName, + (context, input) => ExecuteActivityAsync(binding, input)); + } + } + + // Register agent entities for any AI agents used in workflows + foreach (string agentName in agentFactories.Keys) + { + registry.AddEntity(AgentSessionId.ToEntityName(agentName)); + } + }); + }); + + // Register DurableAgentsOptions and agent factories for entity resolution + if (agentFactories.Count > 0) + { + services.AddSingleton(durableOptions.Agents); + + // Register the agent factories dictionary for backward compatibility + services.TryAddSingleton( + sp => sp.GetRequiredService().GetAgentFactories()); + + // A custom data converter is needed for proper JSON serialization + services.TryAddSingleton(); + } + + // Configure Durable Task Client if a builder is provided + if (clientBuilder is not null) + { + services.AddDurableTaskClient(clientBuilder); + } + + return services; + } + + private static WorkflowRegistrationInfo BuildWorkflowRegistration( + Workflow workflow, + HashSet registeredActivities) + { + string workflowName = workflow.Name!; + string orchestrationName = WorkflowNamingHelper.ToOrchestrationFunctionName(workflowName); + + // Get all executor IDs from the workflow + HashSet executorIds = GetAllExecutorIds(workflow); + Dictionary executorBindings = workflow.ReflectExecutors(); + + List activities = []; + + foreach (string executorId in executorIds) + { + if (!executorBindings.TryGetValue(executorId, out ExecutorBinding? binding)) + { + continue; + } + + string executorName = WorkflowNamingHelper.GetExecutorName(executorId); + string activityName = WorkflowNamingHelper.ToOrchestrationFunctionName(executorName); + + // Skip if already registered (same executor used in multiple workflows) + if (!registeredActivities.Add(activityName)) + { + continue; + } + + // Skip agent executors - they're handled differently + if (binding is AIAgentBinding) + { + continue; + } + + activities.Add(new ActivityRegistrationInfo(activityName, binding)); + } + + return new WorkflowRegistrationInfo(orchestrationName, activities); + } + + private static HashSet GetAllExecutorIds(Workflow workflow) + { + HashSet executorIds = [workflow.StartExecutorId]; + + foreach (KeyValuePair> edgeGroup in workflow.ReflectEdges()) + { + executorIds.Add(edgeGroup.Key); + foreach (EdgeInfo edge in edgeGroup.Value) + { + foreach (string sinkId in edge.Connection.SinkIds) + { + executorIds.Add(sinkId); + } + } + } + + return executorIds; + } + + private static async Task RunWorkflowOrchestrationAsync( + TaskOrchestrationContext context, + string input, + DurableOptions durableOptions) + { + ILogger logger = context.CreateReplaySafeLogger("WorkflowOrchestration"); + DurableWorkflowRunner runner = new( + NullLoggerFactory.Instance.CreateLogger(), + durableOptions); + + return await runner.RunWorkflowOrchestrationAsync(context, input, logger).ConfigureAwait(true); + } + + [UnconditionalSuppressMessage("Trimming", "IL2026:RequiresUnreferencedCode", Justification = "Executor types are registered at startup.")] + [UnconditionalSuppressMessage("AOT", "IL3050:RequiresDynamicCode", Justification = "Executor types are registered at startup.")] + private static async Task ExecuteActivityAsync(ExecutorBinding binding, string input) + { + // Deserialize the input wrapper that includes state + DurableActivityInput? inputWithState = TryDeserializeActivityInput(input); + string executorInput = inputWithState?.Input ?? input; + Dictionary sharedState = inputWithState?.State ?? []; + + // Create executor instance from binding + Executor executor = await binding.FactoryAsync!("activity-run").ConfigureAwait(false); + + Type inputType = executor.InputTypes.FirstOrDefault() ?? typeof(string); + object typedInput = DeserializeInput(executorInput, inputType); + + // Create a pipeline context that has access to shared state and executor + PipelineActivityContext workflowContext = new(sharedState, executor); + + object? result = await executor.ExecuteAsync( + typedInput, + new TypeId(inputType), + workflowContext, + CancellationToken.None).ConfigureAwait(false); + + // Always return wrapped output with state updates, events, and result + DurableActivityOutput output = new() + { + Result = SerializeResult(result), + StateUpdates = workflowContext.StateUpdates, + ClearedScopes = [.. workflowContext.ClearedScopes], + Events = workflowContext.Events.ConvertAll(SerializeEvent) + }; + + return JsonSerializer.Serialize(output); + } + + [UnconditionalSuppressMessage("Trimming", "IL2026:RequiresUnreferencedCode", Justification = "Deserializing known wrapper type.")] + [UnconditionalSuppressMessage("AOT", "IL3050:RequiresDynamicCode", Justification = "Deserializing known wrapper type.")] + private static DurableActivityInput? TryDeserializeActivityInput(string input) + { + try + { + return JsonSerializer.Deserialize(input); + } + catch (JsonException) + { + return null; + } + } + + [UnconditionalSuppressMessage("Trimming", "IL2026:RequiresUnreferencedCode", Justification = "Serializing workflow event types.")] + [UnconditionalSuppressMessage("AOT", "IL3050:RequiresDynamicCode", Justification = "Serializing workflow event types.")] + private static string SerializeEvent(WorkflowEvent evt) + { + // Serialize with type information so we can deserialize to the correct type later + DurableWorkflowRunner.SerializedWorkflowEvent wrapper = new() + { + TypeName = evt.GetType().AssemblyQualifiedName, + Data = JsonSerializer.Serialize(evt, evt.GetType()) + }; + return JsonSerializer.Serialize(wrapper); + } + + [UnconditionalSuppressMessage("Trimming", "IL2026:RequiresUnreferencedCode", Justification = "Deserializing workflow types registered at startup.")] + [UnconditionalSuppressMessage("AOT", "IL3050:RequiresDynamicCode", Justification = "Deserializing workflow types registered at startup.")] + private static object DeserializeInput(string input, Type targetType) + { + if (targetType == typeof(string)) + { + return input; + } + + return JsonSerializer.Deserialize(input, targetType) + ?? throw new InvalidOperationException($"Failed to deserialize input to type '{targetType.Name}'."); + } + + [UnconditionalSuppressMessage("Trimming", "IL2026:RequiresUnreferencedCode", Justification = "Serializing workflow types registered at startup.")] + [UnconditionalSuppressMessage("AOT", "IL3050:RequiresDynamicCode", 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; + } + + return JsonSerializer.Serialize(result, result.GetType()); + } + + private sealed record WorkflowRegistrationInfo(string OrchestrationName, List Activities); + + private sealed record ActivityRegistrationInfo(string ActivityName, ExecutorBinding Binding); + + /// + /// Custom data converter for workflow execution with AI agents. + /// + private sealed class WorkflowDataConverter : DataConverter + { + private static readonly JsonSerializerOptions s_options = new(DurableAgentJsonUtilities.DefaultOptions) + { + PropertyNamingPolicy = JsonNamingPolicy.CamelCase, + PropertyNameCaseInsensitive = true, + }; + + [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Fallback path uses reflection when metadata unavailable.")] + [UnconditionalSuppressMessage("ReflectionAnalysis", "IL3050", Justification = "Fallback path uses reflection when metadata unavailable.")] + public override object? Deserialize(string? data, Type targetType) + { + if (data is null) + { + return null; + } + + if (targetType == typeof(DurableAgentState)) + { + return JsonSerializer.Deserialize(data, DurableAgentStateJsonContext.Default.DurableAgentState); + } + + JsonTypeInfo? typeInfo = s_options.GetTypeInfo(targetType); + if (typeInfo is JsonTypeInfo typedInfo) + { + return JsonSerializer.Deserialize(data, typedInfo); + } + + return JsonSerializer.Deserialize(data, targetType, s_options); + } + + [return: NotNullIfNotNull(nameof(value))] + [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Fallback path uses reflection when metadata unavailable.")] + [UnconditionalSuppressMessage("ReflectionAnalysis", "IL3050", Justification = "Fallback path uses reflection when metadata unavailable.")] + public override string? Serialize(object? value) + { + if (value is null) + { + return null; + } + + if (value is DurableAgentState durableAgentState) + { + return JsonSerializer.Serialize(durableAgentState, DurableAgentStateJsonContext.Default.DurableAgentState); + } + + JsonTypeInfo? typeInfo = s_options.GetTypeInfo(value.GetType()); + if (typeInfo is JsonTypeInfo typedInfo) + { + return JsonSerializer.Serialize(value, typedInfo); + } + + return JsonSerializer.Serialize(value, s_options); + } + } +} + +/// +/// Input payload for activity execution, containing the executor input and shared workflow state. +/// +internal sealed class DurableActivityInput +{ + /// + /// Gets or sets the serialized executor input. + /// + public string? Input { get; set; } + + /// + /// Gets or sets the shared state dictionary (scope-prefixed key -> serialized value). + /// + public Dictionary State { get; set; } = []; +} + +/// +/// Output payload from activity execution, containing the result, state updates, and emitted events. +/// +internal sealed class DurableActivityOutput +{ + /// + /// Gets or sets the serialized result of the activity. + /// + public string? Result { get; set; } + + /// + /// Gets or sets state updates made during activity execution (scope-prefixed key -> serialized value, null = delete). + /// + public Dictionary StateUpdates { get; set; } = []; + + /// + /// Gets or sets scopes that were cleared during activity execution. + /// + public List ClearedScopes { get; set; } = []; + + /// + /// Gets or sets the serialized workflow events emitted during activity execution. + /// + public List Events { get; set; } = []; +} diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/Logs.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/Logs.cs index 4540bd28e6..50c95140ab 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/Logs.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/Logs.cs @@ -136,4 +136,16 @@ internal static partial class Logs Level = LogLevel.Debug, Message = "Executor '{ExecutorId}' skipped due to edge condition evaluation")] public static partial void LogExecutorSkipped(this ILogger logger, string executorId); + + [LoggerMessage( + EventId = 18, + Level = LogLevel.Debug, + Message = "Waiting for external event '{EventName}' with input: {Input}")] + public static partial void LogWaitingForExternalEvent(this ILogger logger, string eventName, string input); + + [LoggerMessage( + EventId = 19, + Level = LogLevel.Debug, + Message = "Received external event '{EventName}' with response: {Response}")] + public static partial void LogReceivedExternalEvent(this ILogger logger, string eventName, string response); } diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/PipelineActivityContext.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/PipelineActivityContext.cs new file mode 100644 index 0000000000..8ccfffde97 --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/PipelineActivityContext.cs @@ -0,0 +1,246 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Diagnostics.CodeAnalysis; +using System.Text.Json; +using Microsoft.Agents.AI.Workflows; + +namespace Microsoft.Agents.AI.DurableTask; + +/// +/// A workflow context for activity execution that uses pipeline-based state management. +/// State is passed in from the orchestration and updates are collected for return. +/// +internal sealed class PipelineActivityContext : IWorkflowContext +{ + private readonly Dictionary _initialState; + private readonly Executor _executor; + + /// + /// Initializes a new instance of the class. + /// + /// The shared state passed from the orchestration. + /// The executor running in this context. + public PipelineActivityContext(Dictionary? initialState, Executor executor) + { + this._initialState = initialState ?? []; + this._executor = executor; + } + + /// + /// Gets the events that were added during activity execution. + /// + public List Events { get; } = []; + + /// + /// Gets the state updates made during activity execution. + /// + public Dictionary StateUpdates { get; } = []; + + /// + /// Gets the scopes that were cleared during activity execution. + /// + public HashSet ClearedScopes { get; } = []; + + /// + public ValueTask AddEventAsync(WorkflowEvent workflowEvent, CancellationToken cancellationToken = default) + { + if (workflowEvent is not null) + { + this.Events.Add(workflowEvent); + } + + return default; + } + + /// + public ValueTask SendMessageAsync(object message, string? targetId = null, CancellationToken cancellationToken = default) => default; + + /// + public ValueTask YieldOutputAsync(object output, CancellationToken cancellationToken = default) + { + if (output is not null) + { + // Validate output type matches executor's declared output types (same as in-process execution) + if (!CanOutput(this._executor.OutputTypes, output.GetType())) + { + throw new InvalidOperationException( + $"Cannot output object of type {output.GetType().Name}. " + + $"Expecting one of [{string.Join(", ", this._executor.OutputTypes)}]."); + } + + this.Events.Add(new DurableYieldedOutputEvent(this._executor.Id, output)); + } + + return default; + } + + /// + public ValueTask RequestHaltAsync() + { + this.Events.Add(new DurableHaltRequestedEvent(this._executor.Id)); + return default; + } + + /// + [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow state types.")] + [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing workflow state types.")] + public ValueTask ReadStateAsync(string key, string? scopeName = null, CancellationToken cancellationToken = default) + { + string scopeKey = GetScopeKey(scopeName, key); + string normalizedScope = scopeName ?? "__default__"; + + // Check if scope was cleared + if (this.ClearedScopes.Contains(normalizedScope)) + { + // Only return from updates made after clear + if (this.StateUpdates.TryGetValue(scopeKey, out string? updatedAfterClear) && updatedAfterClear is not null) + { + return ValueTask.FromResult(JsonSerializer.Deserialize(updatedAfterClear)); + } + + return ValueTask.FromResult(default); + } + + // Check local updates first (read-your-writes) + if (this.StateUpdates.TryGetValue(scopeKey, out string? updated)) + { + if (updated is null) + { + return ValueTask.FromResult(default); + } + + return ValueTask.FromResult(JsonSerializer.Deserialize(updated)); + } + + // Fall back to initial state passed from orchestration + if (this._initialState.TryGetValue(scopeKey, out string? initial)) + { + return ValueTask.FromResult(JsonSerializer.Deserialize(initial)); + } + + return ValueTask.FromResult(default); + } + + /// + [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow state types.")] + [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing workflow state types.")] + 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 + T initialValue = initialStateFactory(); + await this.QueueStateUpdateAsync(key, initialValue, scopeName, cancellationToken).ConfigureAwait(false); + return initialValue; + } + + /// + public ValueTask> ReadStateKeysAsync(string? scopeName = null, CancellationToken cancellationToken = default) + { + string scopePrefix = GetScopePrefix(scopeName); + string normalizedScope = scopeName ?? "__default__"; + HashSet keys = []; + + // If scope was cleared, only return keys from updates made after clear + if (this.ClearedScopes.Contains(normalizedScope)) + { + foreach (KeyValuePair update in this.StateUpdates) + { + if (update.Key.StartsWith(scopePrefix, StringComparison.Ordinal) && update.Value is not null) + { + keys.Add(update.Key[scopePrefix.Length..]); + } + } + + return ValueTask.FromResult(keys); + } + + // Start with keys from initial state + foreach (string stateKey in this._initialState.Keys) + { + if (stateKey.StartsWith(scopePrefix, StringComparison.Ordinal)) + { + keys.Add(stateKey[scopePrefix.Length..]); + } + } + + // Merge with updates + foreach (KeyValuePair update in this.StateUpdates) + { + if (update.Key.StartsWith(scopePrefix, StringComparison.Ordinal)) + { + string foundKey = update.Key[scopePrefix.Length..]; + if (update.Value is not null) + { + keys.Add(foundKey); + } + else + { + keys.Remove(foundKey); + } + } + } + + return ValueTask.FromResult(keys); + } + + /// + [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Serializing workflow state types.")] + [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Serializing workflow state types.")] + public ValueTask QueueStateUpdateAsync(string key, T? value, string? scopeName = null, CancellationToken cancellationToken = default) + { + string scopeKey = GetScopeKey(scopeName, key); + this.StateUpdates[scopeKey] = value is null ? null : JsonSerializer.Serialize(value); + return default; + } + + /// + public ValueTask QueueClearScopeAsync(string? scopeName = null, CancellationToken cancellationToken = default) + { + string normalizedScope = scopeName ?? "__default__"; + this.ClearedScopes.Add(normalizedScope); + + // Remove any pending updates in this scope + string scopePrefix = GetScopePrefix(scopeName); + List keysToRemove = this.StateUpdates.Keys + .Where(k => k.StartsWith(scopePrefix, StringComparison.Ordinal)) + .ToList(); + + foreach (string key in keysToRemove) + { + this.StateUpdates.Remove(key); + } + + return default; + } + + /// + public IReadOnlyDictionary? TraceContext => null; + + /// + public bool ConcurrentRunsEnabled => false; + + private static bool CanOutput(ISet outputTypes, Type messageType) + { + foreach (Type type in outputTypes) + { + if (type.IsAssignableFrom(messageType)) + { + return true; + } + } + + return false; + } + + private static string GetScopeKey(string? scopeName, string key) + => $"{GetScopePrefix(scopeName)}{key}"; + + private static string GetScopePrefix(string? scopeName) + => scopeName is null ? "__default__:" : $"{scopeName}:"; +} diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowExecutionPlan.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowExecutionPlan.cs new file mode 100644 index 0000000000..fc6ed55ad6 --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowExecutionPlan.cs @@ -0,0 +1,45 @@ +// Copyright (c) Microsoft. All rights reserved. + +namespace Microsoft.Agents.AI.DurableTask; + +/// +/// Represents the complete execution plan for a workflow, including parallel execution levels. +/// +public sealed class WorkflowExecutionPlan +{ + /// + /// The execution levels in order. Each level contains executors that can run in parallel. + /// + public List Levels { get; } = []; + + /// + /// Maps each executor ID to its predecessors (for Fan-In result aggregation). + /// + public Dictionary> Predecessors { get; } = []; + + /// + /// Maps each executor ID to its successors (for Fan-Out result distribution). + /// + public Dictionary> Successors { get; } = []; + + /// + /// Maps edge connections (sourceId, targetId) to their condition functions. + /// The condition function takes the predecessor's result and returns true if the edge should be followed. + /// + public Dictionary<(string SourceId, string TargetId), Func?> EdgeConditions { get; } = []; + + /// + /// Maps executor IDs to their output types (for proper deserialization during condition evaluation). + /// + public Dictionary ExecutorOutputTypes { get; } = []; + + /// + /// Gets whether this workflow has any parallel execution opportunities. + /// + public bool HasParallelism => this.Levels.Any(l => l.Executors.Count > 1); + + /// + /// Gets whether this workflow has any Fan-In points. + /// + public bool HasFanIn => this.Levels.Any(l => l.IsFanIn); +} diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowHelper.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowHelper.cs index c9cb307723..fe2da3d28d 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowHelper.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowHelper.cs @@ -10,7 +10,14 @@ namespace Microsoft.Agents.AI.DurableTask; /// /// The unique identifier of the executor. /// Indicates whether this executor is an agentic executor. -public sealed record WorkflowExecutorInfo(string ExecutorId, bool IsAgenticExecutor); +/// The request port if this executor is a request port executor; otherwise, null. +public sealed record WorkflowExecutorInfo(string ExecutorId, bool IsAgenticExecutor, RequestPort? RequestPort = null) +{ + /// + /// Gets a value indicating whether this executor is a request port executor (human-in-the-loop). + /// + public bool IsRequestPortExecutor => this.RequestPort is not null; +} /// /// Represents a level of executors that can be executed in parallel (Fan-Out). @@ -21,48 +28,6 @@ public sealed record WorkflowExecutorInfo(string ExecutorId, bool IsAgenticExecu /// Indicates if this level is a Fan-In point (has executors with multiple predecessors). public sealed record WorkflowExecutionLevel(int Level, List Executors, bool IsFanIn); -/// -/// Represents the complete execution plan for a workflow, including parallel execution levels. -/// -public sealed class WorkflowExecutionPlan -{ - /// - /// The execution levels in order. Each level contains executors that can run in parallel. - /// - public List Levels { get; } = []; - - /// - /// Maps each executor ID to its predecessors (for Fan-In result aggregation). - /// - public Dictionary> Predecessors { get; } = []; - - /// - /// Maps each executor ID to its successors (for Fan-Out result distribution). - /// - public Dictionary> Successors { get; } = []; - - /// - /// Maps edge connections (sourceId, targetId) to their condition functions. - /// The condition function takes the predecessor's result and returns true if the edge should be followed. - /// - public Dictionary<(string SourceId, string TargetId), Func?> EdgeConditions { get; } = []; - - /// - /// Maps executor IDs to their output types (for proper deserialization during condition evaluation). - /// - public Dictionary ExecutorOutputTypes { get; } = []; - - /// - /// Gets whether this workflow has any parallel execution opportunities. - /// - public bool HasParallelism => this.Levels.Any(l => l.Executors.Count > 1); - - /// - /// Gets whether this workflow has any Fan-In points. - /// - public bool HasFanIn => this.Levels.Any(l => l.IsFanIn); -} - /// /// Provides helper methods for analyzing and executing workflows. /// @@ -182,7 +147,8 @@ public static class WorkflowHelper ExecutorBinding executorBinding = executors[executorId]; bool isAgentic = IsAgentExecutorType(executorBinding.ExecutorType); - levelExecutors.Add(new WorkflowExecutorInfo(executorId, isAgentic)); + RequestPort? requestPort = (executorBinding is RequestPortBinding rpb) ? rpb.Port : null; + levelExecutors.Add(new WorkflowExecutorInfo(executorId, isAgentic, requestPort)); // Check Fan-In for this executor if (predecessors[executorId].Count > 1) @@ -215,7 +181,8 @@ public static class WorkflowHelper if (inDegree[executorIndex[executor.Key]] > 0) { bool isAgentic = IsAgentExecutorType(executor.Value.ExecutorType); - remainingExecutors.Add(new WorkflowExecutorInfo(executor.Key, isAgentic)); + RequestPort? requestPort = (executor.Value is RequestPortBinding rpb) ? rpb.Port : null; + remainingExecutors.Add(new WorkflowExecutorInfo(executor.Key, isAgentic, requestPort)); } } diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowSharedStateEntity.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowSharedStateEntity.cs deleted file mode 100644 index 1314edcb31..0000000000 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowSharedStateEntity.cs +++ /dev/null @@ -1,171 +0,0 @@ -// Copyright (c) Microsoft. All rights reserved. - -using Microsoft.DurableTask.Entities; - -namespace Microsoft.Agents.AI.DurableTask; - -/// -/// 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. -/// -public 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. -/// -public 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. -/// -public 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. -/// -public 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; } -} diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableOptionsExtensions.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableOptionsExtensions.cs index b5abd14103..2aab236469 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableOptionsExtensions.cs +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableOptionsExtensions.cs @@ -90,9 +90,6 @@ public static class DurableOptionsExtensions { builder.ConfigureDurableWorker().AddTasks(tasks => { - // Register the workflow state entity for shared state management within workflows. - tasks.AddEntity(WorkflowSharedStateEntity.EntityName); - foreach (string workflowName in workflows.Workflows.Select(kp => kp.Key)) { string orchestrationFunctionName = WorkflowNamingHelper.ToOrchestrationFunctionName(workflowName); diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowRunner.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowRunner.cs index 9b4df91b9e..ab209b54c1 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowRunner.cs +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowRunner.cs @@ -1,6 +1,7 @@ // 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; @@ -30,10 +31,12 @@ internal sealed class FunctionsWorkflowRunner : DurableWorkflowRunner /// Executes an activity function for a workflow executor. /// /// The name of the activity function to execute. - /// The serialized executor input. - /// The durable task client for entity operations. + /// The serialized executor input (may include state via ActivityInputWithState wrapper). + /// The durable task client (unused in pipeline mode, kept for API compatibility). /// The function context containing binding data with the orchestration instance ID. - /// The serialized executor output. + /// The serialized executor output (wrapped in ActivityOutputWithState). + [UnconditionalSuppressMessage("AOT", "IL2026:RequiresUnreferencedCode", Justification = "Executor types are registered at startup.")] + [UnconditionalSuppressMessage("AOT", "IL3050:RequiresDynamicCode", Justification = "Executor types are registered at startup.")] internal async Task ExecuteActivityAsync( string activityFunctionName, string input, @@ -42,7 +45,6 @@ internal sealed class FunctionsWorkflowRunner : DurableWorkflowRunner { ArgumentNullException.ThrowIfNull(activityFunctionName); ArgumentNullException.ThrowIfNull(input); - ArgumentNullException.ThrowIfNull(durableTaskClient); ArgumentNullException.ThrowIfNull(functionContext); string executorName = ParseExecutorName(activityFunctionName); @@ -54,20 +56,19 @@ internal sealed class FunctionsWorkflowRunner : DurableWorkflowRunner this.Logger.LogExecutingActivity(registration.ExecutorId, executorName); + // Deserialize the input wrapper that includes state (pipeline approach) + ActivityInputWithState? inputWithState = TryDeserializeActivityInput(input); + string executorInput = inputWithState?.Input ?? input; + Dictionary sharedState = inputWithState?.State ?? []; + Executor executor = await registration.CreateExecutorInstanceAsync("activity-run", CancellationToken.None) .ConfigureAwait(false); Type inputType = executor.InputTypes.FirstOrDefault() ?? typeof(string); - object typedInput = DeserializeInput(input, inputType); + object typedInput = DeserializeInput(executorInput, 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); + // Create pipeline context that manages state locally with executor ID + FunctionsPipelineActivityContext context = new(sharedState, executor.Id); object? result = await executor.ExecuteAsync( typedInput, @@ -75,26 +76,265 @@ internal sealed class FunctionsWorkflowRunner : DurableWorkflowRunner context, CancellationToken.None).ConfigureAwait(false); - return SerializeResult(result); + // Return wrapped output with state updates, events, and result + ActivityOutputWithState output = new() + { + Result = SerializeResult(result), + StateUpdates = context.StateUpdates, + ClearedScopes = [.. context.ClearedScopes], + Events = context.Events.ConvertAll(SerializeEvent) + }; + + return JsonSerializer.Serialize(output); } - private static string? GetInstanceIdFromContext(FunctionContext functionContext) + [UnconditionalSuppressMessage("AOT", "IL2026:RequiresUnreferencedCode", Justification = "Deserializing known wrapper type.")] + [UnconditionalSuppressMessage("AOT", "IL3050:RequiresDynamicCode", Justification = "Deserializing known wrapper type.")] + private static ActivityInputWithState? TryDeserializeActivityInput(string input) { - if (functionContext.BindingContext.BindingData.TryGetValue("instanceId", out object? instanceIdObj) && - instanceIdObj is string instanceId) + try { - return instanceId; + return JsonSerializer.Deserialize(input); + } + catch (JsonException) + { + return null; + } + } + + [UnconditionalSuppressMessage("AOT", "IL2026:RequiresUnreferencedCode", Justification = "Serializing workflow event types.")] + [UnconditionalSuppressMessage("AOT", "IL3050:RequiresDynamicCode", Justification = "Serializing workflow event types.")] + private static string SerializeEvent(WorkflowEvent evt) + { + // Serialize with type information so we can deserialize to the correct type later +#pragma warning disable IDE0001 // Simplify name - cannot simplify cross-assembly reference + Microsoft.Agents.AI.DurableTask.DurableWorkflowRunner.SerializedWorkflowEvent wrapper = new() +#pragma warning restore IDE0001 + { + TypeName = evt.GetType().AssemblyQualifiedName, + Data = JsonSerializer.Serialize(evt, evt.GetType()) + }; + return JsonSerializer.Serialize(wrapper); + } + + /// + /// A pipeline-based workflow context for Azure Functions activity execution. + /// State is passed in from the orchestration and updates are collected for return. + /// + private sealed class FunctionsPipelineActivityContext : IWorkflowContext + { + private readonly Dictionary _initialState; + private readonly string _executorId; + + public FunctionsPipelineActivityContext(Dictionary? initialState, string executorId) + { + this._initialState = initialState ?? []; + this._executorId = executorId; } - return null; - } + public List Events { get; } = []; + public Dictionary StateUpdates { get; } = []; + public HashSet ClearedScopes { get; } = []; - [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); + public ValueTask AddEventAsync(WorkflowEvent workflowEvent, CancellationToken cancellationToken = default) + { + if (workflowEvent is not null) + { + this.Events.Add(workflowEvent); + } + + return default; + } + + public ValueTask SendMessageAsync(object message, string? targetId = null, CancellationToken cancellationToken = default) => default; + + public ValueTask YieldOutputAsync(object output, CancellationToken cancellationToken = default) + { + // Emit DurableYieldedOutputEvent (WorkflowOutputEvent has internal constructor) + if (output is not null) + { + this.Events.Add(new DurableYieldedOutputEvent(this._executorId, output)); + } + + return default; + } + + public ValueTask RequestHaltAsync() + { + // Emit DurableHaltRequestedEvent (RequestHaltEvent is internal) + this.Events.Add(new DurableHaltRequestedEvent(this._executorId)); + return default; + } + + [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow state types.")] + [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing workflow state types.")] + public ValueTask ReadStateAsync(string key, string? scopeName = null, CancellationToken cancellationToken = default) + { + string scopeKey = GetScopeKey(scopeName, key); + string normalizedScope = scopeName ?? "__default__"; + + if (this.ClearedScopes.Contains(normalizedScope)) + { + if (this.StateUpdates.TryGetValue(scopeKey, out string? updatedAfterClear) && updatedAfterClear is not null) + { + return ValueTask.FromResult(JsonSerializer.Deserialize(updatedAfterClear)); + } + + return ValueTask.FromResult(default); + } + + if (this.StateUpdates.TryGetValue(scopeKey, out string? updated)) + { + return updated is null + ? ValueTask.FromResult(default) + : ValueTask.FromResult(JsonSerializer.Deserialize(updated)); + } + + if (this._initialState.TryGetValue(scopeKey, out string? initial)) + { + return ValueTask.FromResult(JsonSerializer.Deserialize(initial)); + } + + return ValueTask.FromResult(default); + } + + [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow state types.")] + [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing workflow state types.")] + 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; + } + + T initialValue = initialStateFactory(); + await this.QueueStateUpdateAsync(key, initialValue, scopeName, cancellationToken).ConfigureAwait(false); + return initialValue; + } + + public ValueTask> ReadStateKeysAsync(string? scopeName = null, CancellationToken cancellationToken = default) + { + string scopePrefix = GetScopePrefix(scopeName); + string normalizedScope = scopeName ?? "__default__"; + HashSet keys = []; + + if (this.ClearedScopes.Contains(normalizedScope)) + { + foreach (KeyValuePair update in this.StateUpdates) + { + if (update.Key.StartsWith(scopePrefix, StringComparison.Ordinal) && update.Value is not null) + { + keys.Add(update.Key[scopePrefix.Length..]); + } + } + + return ValueTask.FromResult(keys); + } + + foreach (string stateKey in this._initialState.Keys) + { + if (stateKey.StartsWith(scopePrefix, StringComparison.Ordinal)) + { + keys.Add(stateKey[scopePrefix.Length..]); + } + } + + foreach (KeyValuePair update in this.StateUpdates) + { + if (update.Key.StartsWith(scopePrefix, StringComparison.Ordinal)) + { + string foundKey = update.Key[scopePrefix.Length..]; + if (update.Value is not null) + { + keys.Add(foundKey); + } + else + { + keys.Remove(foundKey); + } + } + } + + return ValueTask.FromResult(keys); + } + + [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Serializing workflow state types.")] + [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Serializing workflow state types.")] + public ValueTask QueueStateUpdateAsync(string key, T? value, string? scopeName = null, CancellationToken cancellationToken = default) + { + string scopeKey = GetScopeKey(scopeName, key); + this.StateUpdates[scopeKey] = value is null ? null : JsonSerializer.Serialize(value); + return default; + } + + public ValueTask QueueClearScopeAsync(string? scopeName = null, CancellationToken cancellationToken = default) + { + string normalizedScope = scopeName ?? "__default__"; + this.ClearedScopes.Add(normalizedScope); + + string scopePrefix = GetScopePrefix(scopeName); + List keysToRemove = this.StateUpdates.Keys + .Where(k => k.StartsWith(scopePrefix, StringComparison.Ordinal)) + .ToList(); + + foreach (string key in keysToRemove) + { + this.StateUpdates.Remove(key); + } + + return default; + } + + public IReadOnlyDictionary? TraceContext => null; + public bool ConcurrentRunsEnabled => false; + + private static string GetScopeKey(string? scopeName, string key) + => $"{GetScopePrefix(scopeName)}{key}"; + + private static string GetScopePrefix(string? scopeName) + => scopeName is null ? "__default__:" : $"{scopeName}:"; } } + +/// +/// Wrapper for activity input that includes shared state from the orchestration. +/// +internal sealed class ActivityInputWithState +{ + /// + /// Gets or sets the serialized executor input. + /// + public string? Input { get; set; } + + /// + /// Gets or sets the shared state dictionary (scope-prefixed key -> serialized value). + /// + public Dictionary State { get; set; } = []; +} + +/// +/// Wrapper for activity output that includes state updates and events. +/// +internal sealed class ActivityOutputWithState +{ + /// + /// Gets or sets the serialized result of the activity. + /// + public string? Result { get; set; } + + /// + /// Gets or sets state updates made during activity execution. + /// + public Dictionary StateUpdates { get; set; } = []; + + /// + /// Gets or sets scopes that were cleared during activity execution. + /// + public List ClearedScopes { get; set; } = []; + + /// + /// Gets or sets the serialized workflow events emitted during activity execution. + /// + public List Events { get; set; } = []; +} diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Workflow.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Workflow.cs index afb44bccba..28edf94943 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Workflow.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Workflow.cs @@ -60,15 +60,6 @@ public class Workflow return conditions; } - /// - /// Gets all executor bindings in the workflow, keyed by their ID. - /// - /// A dictionary mapping executor IDs to their . - public Dictionary ReflectExecutors() - { - return new Dictionary(this.ExecutorBindings); - } - internal Dictionary Ports { get; init; } = []; /// @@ -94,6 +85,15 @@ public class Workflow return new Dictionary(this.ExecutorBindings); } + /// + /// Gets the set of executor IDs that are registered as output sources via . + /// + /// A copy of the output executor IDs set. Modifications do not affect the workflow. + public HashSet ReflectOutputExecutors() + { + return new HashSet(this.OutputExecutors); + } + /// /// Gets the identifier of the starting executor of the workflow. ///