diff --git a/dotnet/agent-framework-dotnet.slnx b/dotnet/agent-framework-dotnet.slnx index 31816b9571..b203ed6345 100644 --- a/dotnet/agent-framework-dotnet.slnx +++ b/dotnet/agent-framework-dotnet.slnx @@ -51,6 +51,7 @@ + diff --git a/dotnet/samples/DurableAgents/ConsoleApps/09_SubWorkflows/09_SubWorkflows.csproj b/dotnet/samples/DurableAgents/ConsoleApps/09_SubWorkflows/09_SubWorkflows.csproj new file mode 100644 index 0000000000..904a36f129 --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/09_SubWorkflows/09_SubWorkflows.csproj @@ -0,0 +1,29 @@ + + + net10.0 + Exe + enable + enable + SubWorkflows + SubWorkflows + + + + + + + + + + + + + + + + diff --git a/dotnet/samples/DurableAgents/ConsoleApps/09_SubWorkflows/Executors.cs b/dotnet/samples/DurableAgents/ConsoleApps/09_SubWorkflows/Executors.cs new file mode 100644 index 0000000000..a56bf73a70 --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/09_SubWorkflows/Executors.cs @@ -0,0 +1,218 @@ +// Copyright (c) Microsoft. All rights reserved. + +using Microsoft.Agents.AI.Workflows; + +namespace SubWorkflows; + +// ============================================ +// Order Processing Models +// ============================================ + +/// +/// Represents an order being processed. +/// +internal sealed class OrderInfo +{ + public required string OrderId { get; set; } + public decimal Amount { get; set; } + public string? PaymentTransactionId { get; set; } + public string? InventoryReservationId { get; set; } + public string? TrackingNumber { get; set; } + public string? Carrier { get; set; } +} + +// ============================================ +// Main Workflow Executors +// ============================================ + +/// +/// Entry point executor that receives the order ID and creates an OrderInfo object. +/// +internal sealed class OrderReceived() : Executor("OrderReceived") +{ + public override ValueTask HandleAsync(string message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + Console.WriteLine(); + Console.ForegroundColor = ConsoleColor.Cyan; + Console.WriteLine($"[OrderReceived] Processing order '{message}'"); + Console.ResetColor(); + + OrderInfo order = new() + { + OrderId = message, + Amount = 99.99m // Simulated order amount + }; + + return ValueTask.FromResult(order); + } +} + +/// +/// Final executor that outputs the completed order summary. +/// +internal sealed class OrderCompleted() : Executor("OrderCompleted") +{ + public override ValueTask HandleAsync(OrderInfo message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + Console.WriteLine(); + Console.ForegroundColor = ConsoleColor.Green; + Console.WriteLine("┌─────────────────────────────────────────────────────────────────┐"); + Console.WriteLine($"│ [OrderCompleted] Order '{message.OrderId}' successfully processed!"); + Console.WriteLine($"│ Payment: {message.PaymentTransactionId}"); + Console.WriteLine($"│ Inventory: {message.InventoryReservationId}"); + Console.WriteLine($"│ Shipping: {message.Carrier} - {message.TrackingNumber}"); + Console.WriteLine("└─────────────────────────────────────────────────────────────────┘"); + Console.ResetColor(); + + return ValueTask.FromResult($"Order {message.OrderId} completed. Tracking: {message.TrackingNumber}"); + } +} + +// ============================================ +// Payment Sub-Workflow Executors +// ============================================ + +/// +/// Validates payment information for an order. +/// +internal sealed class ValidatePayment() : Executor("ValidatePayment") +{ + public override async ValueTask HandleAsync(OrderInfo message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + Console.WriteLine(); + Console.ForegroundColor = ConsoleColor.Yellow; + Console.WriteLine($" [Payment/ValidatePayment] Validating payment for order '{message.OrderId}'..."); + Console.ResetColor(); + + await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken); + + Console.ForegroundColor = ConsoleColor.Yellow; + Console.WriteLine($" [Payment/ValidatePayment] Payment validated for ${message.Amount}"); + Console.ResetColor(); + + return message; + } +} + +/// +/// Charges the payment for an order. +/// +internal sealed class ChargePayment() : Executor("ChargePayment") +{ + public override async ValueTask HandleAsync(OrderInfo message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + Console.ForegroundColor = ConsoleColor.Yellow; + Console.WriteLine($" [Payment/ChargePayment] Charging ${message.Amount} for order '{message.OrderId}'..."); + Console.ResetColor(); + + await Task.Delay(TimeSpan.FromSeconds(2), cancellationToken); + + message.PaymentTransactionId = $"TXN-{Guid.NewGuid().ToString("N")[..8].ToUpperInvariant()}"; + + Console.ForegroundColor = ConsoleColor.Yellow; + Console.WriteLine($" [Payment/ChargePayment] ✓ Payment processed: {message.PaymentTransactionId}"); + Console.ResetColor(); + + return message; + } +} + +// ============================================ +// Inventory Sub-Workflow Executors +// ============================================ + +/// +/// Checks inventory availability for an order. +/// +internal sealed class CheckInventory() : Executor("CheckInventory") +{ + public override async ValueTask HandleAsync(OrderInfo message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + Console.WriteLine(); + Console.ForegroundColor = ConsoleColor.Magenta; + Console.WriteLine($" [Inventory/CheckInventory] Checking inventory for order '{message.OrderId}'..."); + Console.ResetColor(); + + await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken); + + Console.ForegroundColor = ConsoleColor.Magenta; + Console.WriteLine(" [Inventory/CheckInventory] ✓ Items available in stock"); + Console.ResetColor(); + + return message; + } +} + +/// +/// Reserves inventory for an order. +/// +internal sealed class ReserveInventory() : Executor("ReserveInventory") +{ + public override async ValueTask HandleAsync(OrderInfo message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + Console.ForegroundColor = ConsoleColor.Magenta; + Console.WriteLine($" [Inventory/ReserveInventory] Reserving items for order '{message.OrderId}'..."); + Console.ResetColor(); + + await Task.Delay(TimeSpan.FromSeconds(2), cancellationToken); + + message.InventoryReservationId = $"RES-{Guid.NewGuid().ToString("N")[..8].ToUpperInvariant()}"; + + Console.ForegroundColor = ConsoleColor.Magenta; + Console.WriteLine($" [Inventory/ReserveInventory] ✓ Reserved: {message.InventoryReservationId}"); + Console.ResetColor(); + + return message; + } +} + +// ============================================ +// Shipping Sub-Workflow Executors +// ============================================ + +/// +/// Selects a shipping carrier for an order. +/// +internal sealed class SelectCarrier() : Executor("SelectCarrier") +{ + public override async ValueTask HandleAsync(OrderInfo message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + Console.WriteLine(); + Console.ForegroundColor = ConsoleColor.Blue; + Console.WriteLine($" [Shipping/SelectCarrier] Selecting carrier for order '{message.OrderId}'..."); + Console.ResetColor(); + + await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken); + + message.Carrier = message.Amount > 50 ? "Express" : "Standard"; + + Console.ForegroundColor = ConsoleColor.Blue; + Console.WriteLine($" [Shipping/SelectCarrier] ✓ Selected carrier: {message.Carrier}"); + Console.ResetColor(); + + return message; + } +} + +/// +/// Creates shipment and generates tracking number. +/// +internal sealed class CreateShipment() : Executor("CreateShipment") +{ + public override async ValueTask HandleAsync(OrderInfo message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + Console.ForegroundColor = ConsoleColor.Blue; + Console.WriteLine($" [Shipping/CreateShipment] Creating shipment for order '{message.OrderId}'..."); + Console.ResetColor(); + + await Task.Delay(TimeSpan.FromSeconds(2), cancellationToken); + + message.TrackingNumber = $"TRACK-{Guid.NewGuid().ToString("N")[..10].ToUpperInvariant()}"; + + Console.ForegroundColor = ConsoleColor.Blue; + Console.WriteLine($" [Shipping/CreateShipment] ✓ Shipment created: {message.TrackingNumber}"); + Console.ResetColor(); + + return message; + } +} diff --git a/dotnet/samples/DurableAgents/ConsoleApps/09_SubWorkflows/Program.cs b/dotnet/samples/DurableAgents/ConsoleApps/09_SubWorkflows/Program.cs new file mode 100644 index 0000000000..8fed264c2b --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/09_SubWorkflows/Program.cs @@ -0,0 +1,175 @@ +// Copyright (c) Microsoft. All rights reserved. + +// This sample demonstrates how to use sub-workflows within a durable orchestration. +// Sub-workflows allow you to compose complex workflows from simpler, reusable components. +// +// The sample implements an order processing workflow with three sub-workflows: +// 1. PaymentProcessing - Validates and processes payment +// 2. InventoryManagement - Checks and reserves inventory +// 3. ShippingArrangement - Arranges shipping and generates tracking +// +// Each sub-workflow runs as a separate orchestration instance, visible in the DTS dashboard. +// This provides: +// - Modular, reusable workflow components +// - Independent checkpointing and replay +// - Hierarchical visualization in the dashboard +// - Failure isolation between parent and child workflows + +using Microsoft.Agents.AI.DurableTask; +using Microsoft.Agents.AI.Workflows; +using Microsoft.DurableTask.Client.AzureManaged; +using Microsoft.DurableTask.Worker.AzureManaged; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using SubWorkflows; + +// Get DTS connection string from environment variable +string dtsConnectionString = Environment.GetEnvironmentVariable("DURABLE_TASK_SCHEDULER_CONNECTION_STRING") + ?? "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None"; + +// ============================================ +// Step 1: Build the Payment Processing sub-workflow +// ============================================ +ValidatePayment validatePayment = new(); +ChargePayment chargePayment = new(); + +Workflow paymentWorkflow = new WorkflowBuilder(validatePayment) + .WithName("SubPaymentProcessing") + .WithDescription("Validates and processes payment for an order") + .AddEdge(validatePayment, chargePayment) + .Build(); + +// ============================================ +// Step 2: Build the Inventory Management sub-workflow +// ============================================ +CheckInventory checkInventory = new(); +ReserveInventory reserveInventory = new(); + +Workflow inventoryWorkflow = new WorkflowBuilder(checkInventory) + .WithName("SubInventoryManagement") + .WithDescription("Checks availability and reserves inventory") + .AddEdge(checkInventory, reserveInventory) + .Build(); + +// ============================================ +// Step 3: Build the Shipping Arrangement sub-workflow +// ============================================ +SelectCarrier selectCarrier = new(); +CreateShipment createShipment = new(); + +Workflow shippingWorkflow = new WorkflowBuilder(selectCarrier) + .WithName("SubShippingArrangement") + .WithDescription("Selects carrier and creates shipment") + .AddEdge(selectCarrier, createShipment) + .Build(); + +// ============================================ +// Step 4: Build the Main Order Processing workflow using sub-workflows +// ============================================ +// Bind sub-workflows as executors for use in the main workflow +ExecutorBinding paymentExecutor = paymentWorkflow.BindAsExecutor("Payment"); +ExecutorBinding inventoryExecutor = inventoryWorkflow.BindAsExecutor("Inventory"); +ExecutorBinding shippingExecutor = shippingWorkflow.BindAsExecutor("Shipping"); + +// Create entry and exit executors for the main workflow +OrderReceived orderReceived = new(); +OrderCompleted orderCompleted = new(); + +// Build the main workflow: OrderReceived -> Payment -> Inventory -> Shipping -> OrderCompleted +Workflow orderProcessingWorkflow = new WorkflowBuilder(orderReceived) + .WithName("OrderProcessing") + .WithDescription("Processes an order through payment, inventory, and shipping") + .AddEdge(orderReceived, paymentExecutor) + .AddEdge(paymentExecutor, inventoryExecutor) + .AddEdge(inventoryExecutor, shippingExecutor) + .AddEdge(shippingExecutor, orderCompleted) + .Build(); + +// ============================================ +// Step 5: Configure and start the host +// ============================================ +IHost host = Host.CreateDefaultBuilder(args) + .ConfigureLogging(logging => logging.SetMinimumLevel(LogLevel.Warning)) + .ConfigureServices(services => + { + // Register only the main workflow - sub-workflows are discovered automatically! + services.ConfigureDurableWorkflows( + options => options.Workflows.AddWorkflow(orderProcessingWorkflow), + workerBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString), + clientBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString)); + }) + .Build(); + +await host.StartAsync(); + +// Get the IWorkflowClient from DI +IWorkflowClient workflowClient = host.Services.GetRequiredService(); + +Console.WriteLine("╔══════════════════════════════════════════════════════════════════╗"); +Console.WriteLine("║ Durable Sub-Workflows Sample ║"); +Console.WriteLine("╠══════════════════════════════════════════════════════════════════╣"); +Console.WriteLine("║ Main Workflow: OrderProcessing ║"); +Console.WriteLine("║ ├── Payment (sub-workflow) ║"); +Console.WriteLine("║ │ ├── ValidatePayment (1s) ║"); +Console.WriteLine("║ │ └── ChargePayment (2s) ║"); +Console.WriteLine("║ ├── Inventory (sub-workflow) ║"); +Console.WriteLine("║ │ ├── CheckInventory (1s) ║"); +Console.WriteLine("║ │ └── ReserveInventory (2s) ║"); +Console.WriteLine("║ └── Shipping (sub-workflow) ║"); +Console.WriteLine("║ ├── SelectCarrier (1s) ║"); +Console.WriteLine("║ └── CreateShipment (2s) ║"); +Console.WriteLine("╚══════════════════════════════════════════════════════════════════╝"); +Console.WriteLine(); +Console.WriteLine("Open the DTS dashboard at http://localhost:8080 to see the"); +Console.WriteLine("parent-child orchestration hierarchy in the Timeline view!"); +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, orderProcessingWorkflow, workflowClient); + } + catch (Exception ex) + { + Console.WriteLine($"Error: {ex.Message}"); + } + + Console.WriteLine(); +} + +await host.StopAsync(); + +// Start a new workflow using IWorkflowClient +async Task StartNewWorkflowAsync(string orderId, Workflow workflow, IWorkflowClient client) +{ + Console.WriteLine($"\nStarting order processing for '{orderId}'..."); + + await using DurableRun run = (DurableRun)await client.RunAsync(workflow, orderId); + Console.WriteLine($"Instance ID: {run.InstanceId}"); + Console.WriteLine("Check the DTS dashboard Timeline tab to see sub-orchestrations!"); + Console.WriteLine(); + + try + { + string? result = await run.WaitForCompletionAsync(); + Console.ForegroundColor = ConsoleColor.Green; + Console.WriteLine($"✓ Order completed: {result}"); + Console.ResetColor(); + } + catch (InvalidOperationException ex) + { + Console.ForegroundColor = ConsoleColor.Red; + Console.WriteLine($"✗ Failed: {ex.Message}"); + Console.ResetColor(); + } +} diff --git a/dotnet/samples/DurableAgents/ConsoleApps/09_SubWorkflows/README.md b/dotnet/samples/DurableAgents/ConsoleApps/09_SubWorkflows/README.md new file mode 100644 index 0000000000..6be9492867 --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/09_SubWorkflows/README.md @@ -0,0 +1,153 @@ +# Sub-Workflows Console Sample + +This sample demonstrates how to compose workflows hierarchically by using sub-workflows within a durable orchestration. Sub-workflows are executed as separate orchestration instances, providing modularity, reusability, and excellent visibility in the Durable Task dashboard. + +## Overview + +The sample implements an order processing system with three sub-workflows: + +``` +OrderProcessing (Main Workflow) +??? OrderReceived +??? Payment (Sub-Workflow) +? ??? ValidatePayment (1s) +? ??? ChargePayment (2s) +??? Inventory (Sub-Workflow) +? ??? CheckInventory (1s) +? ??? ReserveInventory (2s) +??? Shipping (Sub-Workflow) +? ??? SelectCarrier (1s) +? ??? CreateShipment (2s) +??? OrderCompleted +``` + +## Key Concepts + +### Sub-Workflow Benefits + +1. **Modularity**: Each sub-workflow encapsulates related logic (payment, inventory, shipping) +2. **Reusability**: Sub-workflows can be used in multiple parent workflows +3. **Independent Execution**: Each sub-workflow runs as a separate orchestration instance +4. **Dashboard Visibility**: Sub-workflows appear in the Timeline view with parent-child relationships +5. **Failure Isolation**: A failure in a sub-workflow doesn't corrupt the parent's state + +### How Sub-Workflows Work + +```csharp +// Step 1: Build a sub-workflow +Workflow paymentWorkflow = new WorkflowBuilder(validatePayment) + .WithName("PaymentProcessing") + .AddEdge(validatePayment, chargePayment) + .Build(); + +// Step 2: Bind it as an executor for use in a parent workflow +ExecutorBinding paymentExecutor = paymentWorkflow.BindAsExecutor("Payment"); + +// Step 3: Use the sub-workflow executor in the main workflow +Workflow mainWorkflow = new WorkflowBuilder(orderReceived) + .AddEdge(orderReceived, paymentExecutor) // Sub-workflow as an edge target + .AddEdge(paymentExecutor, inventoryExecutor) + .Build(); + +// Step 4: Register only the main workflow - sub-workflows are discovered automatically! +services.ConfigureDurableWorkflows( + options => options.Workflows.AddWorkflow(mainWorkflow), + ...); +``` + +### Dashboard Visualization + +Open the DTS dashboard at `http://localhost:8080` after running a workflow: + +1. Click on the main orchestration instance +2. Switch to the **Timeline** tab +3. You'll see a hierarchical view showing: + - `OrderProcessing` (parent orchestration) + - `PaymentProcessing` (sub-orchestration) + - `InventoryManagement` (sub-orchestration) + - `ShippingArrangement` (sub-orchestration) + +Each sub-orchestration has its own instance ID and can be inspected independently. + +## Environment Setup + +See the [README.md](../README.md) file in the parent directory for information on: +- Installing prerequisites (.NET 10+, Docker) +- Starting the Durable Task Scheduler emulator +- Configuring environment variables + +## Running the Sample + +```bash +# Start the DTS emulator (if not already running) +docker run -d --name dts-emulator -p 8080:8080 -p 8082:8082 mcr.microsoft.com/dts/dts-emulator:latest + +# Run the sample +cd dotnet/samples/DurableAgents/ConsoleApps/09_SubWorkflows +dotnet run --framework net10.0 +``` + +### Sample Session + +```text +???????????????????????????????????????????????????????????????????? +? Durable Sub-Workflows Sample ? +???????????????????????????????????????????????????????????????????? +? Main Workflow: OrderProcessing ? +? ??? Payment (sub-workflow) ? +? ? ??? ValidatePayment (1s) ? +? ? ??? ChargePayment (2s) ? +? ??? Inventory (sub-workflow) ? +? ? ??? CheckInventory (1s) ? +? ? ??? ReserveInventory (2s) ? +? ??? Shipping (sub-workflow) ? +? ??? SelectCarrier (1s) ? +? ??? CreateShipment (2s) ? +???????????????????????????????????????????????????????????????????? + +Open the DTS dashboard at http://localhost:8080 to see the +parent-child orchestration hierarchy in the Timeline view! + +Enter an order ID (or 'exit'): +> ORD-12345 + +Starting order processing for 'ORD-12345'... +Instance ID: abc123def456 +Check the DTS dashboard Timeline tab to see sub-orchestrations! + +[OrderReceived] Processing order 'ORD-12345' + [Payment/ValidatePayment] Validating payment for order 'ORD-12345'... + [Payment/ValidatePayment] Payment validated for $99.99 + [Payment/ChargePayment] Charging $99.99 for order 'ORD-12345'... + [Payment/ChargePayment] ? Payment processed: TXN-A1B2C3D4 + [Inventory/CheckInventory] Checking inventory for order 'ORD-12345'... + [Inventory/CheckInventory] ? Items available in stock + [Inventory/ReserveInventory] Reserving items for order 'ORD-12345'... + [Inventory/ReserveInventory] ? Reserved: RES-E5F6G7H8 + [Shipping/SelectCarrier] Selecting carrier for order 'ORD-12345'... + [Shipping/SelectCarrier] ? Selected carrier: Express + [Shipping/CreateShipment] Creating shipment for order 'ORD-12345'... + [Shipping/CreateShipment] ? Shipment created: TRACK-I9J0K1L2M3 +??????????????????????????????????????????????????????????????????? +? [OrderCompleted] Order 'ORD-12345' successfully processed! +? Payment: TXN-A1B2C3D4 +? Inventory: RES-E5F6G7H8 +? Shipping: Express - TRACK-I9J0K1L2M3 +??????????????????????????????????????????????????????????????????? +? Order completed. Tracking: TRACK-I9J0K1L2M3 +``` + +## Comparison with In-Process Sub-Workflows + +| Feature | In-Process | Durable | +|---------|------------|---------| +| Execution | Same process, synchronized supersteps | Separate orchestration instances | +| Visibility | Single workflow view | Hierarchical dashboard view | +| Checkpointing | Parent checkpoints include child state | Independent checkpoints per sub-workflow | +| Failure Recovery | Parent must handle child failures | Automatic retry with state preservation | +| Scalability | Single process | Can scale across workers | + +## Related Samples + +- [06_SubWorkflows (In-Process)](../../../GettingStarted/Workflows/_Foundational/06_SubWorkflows) - In-process sub-workflow execution +- [08_SingleWorkflow](../08_SingleWorkflow) - Basic durable workflow without sub-workflows diff --git a/dotnet/samples/DurableAgents/ConsoleApps/README.md b/dotnet/samples/DurableAgents/ConsoleApps/README.md index 1bd2b0d224..5ad0efc708 100644 --- a/dotnet/samples/DurableAgents/ConsoleApps/README.md +++ b/dotnet/samples/DurableAgents/ConsoleApps/README.md @@ -9,6 +9,8 @@ This directory contains samples for console app hosting of durable agents. These - **[05_AgentOrchestration_HITL](05_AgentOrchestration_HITL)**: A sample that demonstrates how to implement a human-in-the-loop workflow using durable orchestration, including interactive approval prompts. - **[06_LongRunningTools](06_LongRunningTools)**: A sample that demonstrates how agents can start and interact with durable orchestrations from tool calls to enable long-running tool scenarios. - **[07_ReliableStreaming](07_ReliableStreaming)**: A sample that demonstrates how to implement reliable streaming for durable agents using Redis Streams, enabling clients to disconnect and reconnect without losing messages. +- **[08_SingleWorkflow](08_SingleWorkflow)**: A sample that demonstrates how to run a simple workflow as a durable orchestration, showcasing activity durability and automatic resume on restart. +- **[09_SubWorkflows](09_SubWorkflows)**: A sample that demonstrates how to compose workflows hierarchically using sub-workflows, which run as separate orchestration instances visible in the DTS dashboard. ## Running the Samples diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunner.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunner.cs index b210d8b7e2..c7073c2be3 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunner.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunner.cs @@ -565,8 +565,8 @@ internal class DurableWorkflowRunner /// A containing metadata about the executor. /// Thrown when the executor ID is not found in bindings. /// - /// This method determines the executor type (agentic vs regular) and extracts request port - /// information for human-in-the-loop executors. + /// This method determines the executor type (agentic, sub-workflow, request port, or regular) + /// and extracts the appropriate metadata for each type. /// private static WorkflowExecutorInfo CreateExecutorInfo( string executorId, @@ -579,8 +579,9 @@ internal class DurableWorkflowRunner bool isAgentic = WorkflowHelper.IsAgentExecutorType(binding.ExecutorType); RequestPort? requestPort = (binding is RequestPortBinding rpb) ? rpb.Port : null; + Workflow? subWorkflow = (binding is SubworkflowBinding swb) ? swb.WorkflowInstance : null; - return new WorkflowExecutorInfo(executorId, isAgentic, requestPort); + return new WorkflowExecutorInfo(executorId, isAgentic, requestPort, subWorkflow); } /// @@ -1092,6 +1093,12 @@ internal class DurableWorkflowRunner /// /// /// + /// Sub-Workflow Executors: Executed as sub-orchestrations. + /// The child workflow runs as a separate orchestration instance with its own instance ID. + /// + /// + /// + /// /// Regular Executors: Invoked as Durable Task activities. Input is wrapped /// with state and type information via . /// @@ -1119,6 +1126,12 @@ internal class DurableWorkflowRunner return await ExecuteRequestPortAsync(context, executorInfo, input, logger, customStatus).ConfigureAwait(true); } + // Handle Sub-Workflow executors by calling as sub-orchestrations + if (executorInfo.IsSubworkflowExecutor) + { + return await ExecuteSubWorkflowAsync(context, executorInfo, input, logger).ConfigureAwait(true); + } + if (!executorInfo.IsAgenticExecutor) { string executorName = WorkflowNamingHelper.GetExecutorName(executorInfo.ExecutorId); @@ -1139,6 +1152,54 @@ internal class DurableWorkflowRunner return await ExecuteAgentAsync(context, executorInfo, input, logger).ConfigureAwait(true); } + /// + /// Executes a sub-workflow as a sub-orchestration. + /// + /// The orchestration context for calling sub-orchestrations. + /// The executor info containing the sub-workflow reference. + /// The input to pass to the sub-workflow. + /// The logger for tracing. + /// The result of the sub-workflow execution. + /// + /// + /// Sub-workflows are executed as separate orchestration instances. + /// This provides: + /// + /// + /// Separate instance ID for the child workflow (visible in dashboard) + /// Independent checkpointing and replay + /// Failure isolation (child failure doesn't corrupt parent state) + /// Hierarchical visualization in the Durable Task dashboard + /// + /// + private static async Task ExecuteSubWorkflowAsync( + TaskOrchestrationContext context, + WorkflowExecutorInfo executorInfo, + string input, + ILogger logger) + { + Workflow subWorkflow = executorInfo.SubWorkflow!; + string subOrchestrationName = WorkflowNamingHelper.ToOrchestrationFunctionName(subWorkflow.Name!); + + logger.LogDebug( + "Calling sub-orchestration '{SubOrchestrationName}' for sub-workflow '{SubWorkflowName}'", + subOrchestrationName, + subWorkflow.Name); + + // Call the sub-workflow as a sub-orchestration + // The Durable Task Framework handles checkpointing, replay, and failure isolation + string result = await context.CallSubOrchestratorAsync( + subOrchestrationName, + input).ConfigureAwait(true); + + logger.LogDebug( + "Sub-orchestration '{SubOrchestrationName}' completed with result length: {ResultLength}", + subOrchestrationName, + result?.Length ?? 0); + + return result ?? string.Empty; + } + /// /// Executes a request port executor by waiting for an external event (human-in-the-loop). /// diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowServiceCollectionExtensions.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowServiceCollectionExtensions.cs index f083e2b42d..4118bc3b12 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowServiceCollectionExtensions.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowServiceCollectionExtensions.cs @@ -49,13 +49,21 @@ public static class DurableWorkflowServiceCollectionExtensions // Register the workflow runner services.AddSingleton(); - // Build registration info for all workflows + // Build registration info for all workflows (including sub-workflows) List registrations = []; HashSet registeredActivities = []; + HashSet registeredOrchestrations = []; - foreach (KeyValuePair workflowEntry in durableOptions.Workflows.Workflows) + // Take a snapshot of the workflows to avoid collection modified during enumeration + // (sub-workflows are added to the collection during recursive registration) + foreach (Workflow workflow in durableOptions.Workflows.Workflows.Values.ToList()) { - registrations.Add(BuildWorkflowRegistration(workflowEntry.Value, registeredActivities)); + BuildWorkflowRegistrationRecursive( + workflow, + durableOptions.Workflows, + registrations, + registeredActivities, + registeredOrchestrations); } // Get any AI agents that were auto-registered from workflows @@ -121,6 +129,53 @@ public static class DurableWorkflowServiceCollectionExtensions return services; } + /// + /// Recursively builds workflow registrations, including any sub-workflows. + /// Also adds sub-workflows to the workflow options so they can be looked up at runtime. + /// + /// The workflow to register. + /// The workflow options to add sub-workflows to. + /// The list to add registrations to. + /// Set of already registered activity names to avoid duplicates. + /// Set of already registered orchestration names to avoid duplicates. + private static void BuildWorkflowRegistrationRecursive( + Workflow workflow, + DurableWorkflowOptions workflowOptions, + List registrations, + HashSet registeredActivities, + HashSet registeredOrchestrations) + { + string orchestrationName = WorkflowNamingHelper.ToOrchestrationFunctionName(workflow.Name!); + + // Skip if this workflow is already registered (handles circular references) + if (!registeredOrchestrations.Add(orchestrationName)) + { + return; + } + + // Build registration for this workflow + registrations.Add(BuildWorkflowRegistration(workflow, registeredActivities)); + + // Recursively register any sub-workflows + foreach (KeyValuePair entry in workflow.ReflectExecutors()) + { + if (entry.Value is SubworkflowBinding subworkflowBinding) + { + Workflow subWorkflow = subworkflowBinding.WorkflowInstance; + + // Add sub-workflow to options so it can be looked up by the runner at runtime + workflowOptions.AddWorkflow(subWorkflow); + + BuildWorkflowRegistrationRecursive( + subWorkflow, + workflowOptions, + registrations, + registeredActivities, + registeredOrchestrations); + } + } + } + private static WorkflowRegistrationInfo BuildWorkflowRegistration( Workflow workflow, HashSet registeredActivities) @@ -139,6 +194,12 @@ public static class DurableWorkflowServiceCollectionExtensions continue; } + // Skip sub-workflow executors - they're handled as sub-orchestrations + if (entry.Value is SubworkflowBinding) + { + continue; + } + string executorName = WorkflowNamingHelper.GetExecutorName(entry.Key); string activityName = WorkflowNamingHelper.ToOrchestrationFunctionName(executorName); @@ -237,7 +298,18 @@ public static class DurableWorkflowServiceCollectionExtensions } // Try to load the type directly (for types not in supported types) - return Type.GetType(inputTypeName) ?? supportedTypes.FirstOrDefault() ?? typeof(string); + Type? loadedType = Type.GetType(inputTypeName); + + // If the loaded type is string but the executor doesn't support string, + // fall back to the first supported type. This handles the case where + // serialized JSON objects are passed with type "System.String" but need + // to be deserialized to the actual expected type (e.g., OrderInfo). + if (loadedType == typeof(string) && !supportedTypes.Contains(typeof(string))) + { + return supportedTypes.FirstOrDefault() ?? typeof(string); + } + + return loadedType ?? supportedTypes.FirstOrDefault() ?? typeof(string); } /// diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowHelper.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowHelper.cs index 4fbb5b2368..22b917daa2 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowHelper.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowHelper.cs @@ -11,13 +11,19 @@ namespace Microsoft.Agents.AI.DurableTask; /// The unique identifier of the executor. /// Indicates whether this executor is an agentic executor. /// The request port if this executor is a request port executor; otherwise, null. -[DebuggerDisplay("{ExecutorId}, Agentic = {IsAgenticExecutor}, HITL = {IsRequestPortExecutor}")] -internal sealed record WorkflowExecutorInfo(string ExecutorId, bool IsAgenticExecutor, RequestPort? RequestPort = null) +/// The sub-workflow if this executor is a sub-workflow executor; otherwise, null. +[DebuggerDisplay("{ExecutorId}, Agentic = {IsAgenticExecutor}, HITL = {IsRequestPortExecutor}, SubWorkflow = {IsSubworkflowExecutor}")] +internal sealed record WorkflowExecutorInfo(string ExecutorId, bool IsAgenticExecutor, RequestPort? RequestPort = null, Workflow? SubWorkflow = 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; + + /// + /// Gets a value indicating whether this executor is a sub-workflow executor. + /// + public bool IsSubworkflowExecutor => this.SubWorkflow is not null; } /// @@ -176,7 +182,8 @@ internal static class WorkflowHelper ExecutorBinding executorBinding = executors[executorId]; bool isAgentic = IsAgentExecutorType(executorBinding.ExecutorType); RequestPort? requestPort = (executorBinding is RequestPortBinding rpb) ? rpb.Port : null; - levelExecutors.Add(new WorkflowExecutorInfo(executorId, isAgentic, requestPort)); + Workflow? subWorkflow = (executorBinding is SubworkflowBinding swb) ? swb.WorkflowInstance : null; + levelExecutors.Add(new WorkflowExecutorInfo(executorId, isAgentic, requestPort, subWorkflow)); // Check Fan-In for this executor (excluding back-edges) int nonBackEdgePredecessors = predecessors[executorId] @@ -218,7 +225,8 @@ internal static class WorkflowHelper { bool isAgentic = IsAgentExecutorType(executor.Value.ExecutorType); RequestPort? requestPort = (executor.Value is RequestPortBinding rpb) ? rpb.Port : null; - remainingExecutors.Add(new WorkflowExecutorInfo(executor.Key, isAgentic, requestPort)); + Workflow? subWorkflow = (executor.Value is SubworkflowBinding swb) ? swb.WorkflowInstance : null; + remainingExecutors.Add(new WorkflowExecutorInfo(executor.Key, isAgentic, requestPort, subWorkflow)); } }