diff --git a/dotnet/samples/04-hosting/DurableWorkflows/AzureFunctions/02_ConcurrentWorkflow/Program.cs b/dotnet/samples/04-hosting/DurableWorkflows/AzureFunctions/02_ConcurrentWorkflow/Program.cs index 6b26143981..6532009d4b 100644 --- a/dotnet/samples/04-hosting/DurableWorkflows/AzureFunctions/02_ConcurrentWorkflow/Program.cs +++ b/dotnet/samples/04-hosting/DurableWorkflows/AzureFunctions/02_ConcurrentWorkflow/Program.cs @@ -34,7 +34,7 @@ AggregatorExecutor aggregator = new(); Workflow workflow = new WorkflowBuilder(parseQuestion) .WithName("ExpertReview") .AddFanOutEdge(parseQuestion, [physicist, chemist]) - .AddFanInEdge([physicist, chemist], aggregator) + .AddFanInBarrierEdge([physicist, chemist], aggregator) .Build(); using IHost app = FunctionsApplication diff --git a/dotnet/samples/04-hosting/DurableWorkflows/AzureFunctions/03_WorkflowHITL/Program.cs b/dotnet/samples/04-hosting/DurableWorkflows/AzureFunctions/03_WorkflowHITL/Program.cs index 65f4b7acba..1aa1972e62 100644 --- a/dotnet/samples/04-hosting/DurableWorkflows/AzureFunctions/03_WorkflowHITL/Program.cs +++ b/dotnet/samples/04-hosting/DurableWorkflows/AzureFunctions/03_WorkflowHITL/Program.cs @@ -40,7 +40,7 @@ Workflow expenseApproval = new WorkflowBuilder(createRequest) .AddEdge(createRequest, managerApproval) .AddEdge(managerApproval, prepareFinanceReview) .AddFanOutEdge(prepareFinanceReview, [budgetApproval, complianceApproval]) - .AddFanInEdge([budgetApproval, complianceApproval], reimburse) + .AddFanInBarrierEdge([budgetApproval, complianceApproval], reimburse) .Build(); using IHost app = FunctionsApplication diff --git a/dotnet/samples/04-hosting/DurableWorkflows/ConsoleApps/02_ConcurrentWorkflow/Program.cs b/dotnet/samples/04-hosting/DurableWorkflows/ConsoleApps/02_ConcurrentWorkflow/Program.cs index 1f51d052ea..ae68a56562 100644 --- a/dotnet/samples/04-hosting/DurableWorkflows/ConsoleApps/02_ConcurrentWorkflow/Program.cs +++ b/dotnet/samples/04-hosting/DurableWorkflows/ConsoleApps/02_ConcurrentWorkflow/Program.cs @@ -56,7 +56,7 @@ AggregatorExecutor aggregator = new(); Workflow workflow = new WorkflowBuilder(parseQuestion) .WithName("ExpertReview") .AddFanOutEdge(parseQuestion, [physicist, chemist]) - .AddFanInEdge([physicist, chemist], aggregator) + .AddFanInBarrierEdge([physicist, chemist], aggregator) .Build(); // Configure and start the host diff --git a/dotnet/samples/04-hosting/DurableWorkflows/ConsoleApps/04_WorkflowAndAgents/Program.cs b/dotnet/samples/04-hosting/DurableWorkflows/ConsoleApps/04_WorkflowAndAgents/Program.cs index e137388cb6..5dfec4f277 100644 --- a/dotnet/samples/04-hosting/DurableWorkflows/ConsoleApps/04_WorkflowAndAgents/Program.cs +++ b/dotnet/samples/04-hosting/DurableWorkflows/ConsoleApps/04_WorkflowAndAgents/Program.cs @@ -54,7 +54,7 @@ Workflow physicsWorkflow = new WorkflowBuilder(questionParser) Workflow expertTeamWorkflow = new WorkflowBuilder(questionParser) .WithName("ExpertTeamReview") .AddFanOutEdge(questionParser, [biologist, physicist]) -.AddFanInEdge([biologist, physicist], responseAggregator) +.AddFanInBarrierEdge([biologist, physicist], responseAggregator) .Build(); Workflow chemistryWorkflow = new WorkflowBuilder(questionParser) @@ -96,12 +96,12 @@ IWorkflowClient workflowClient = services.GetRequiredService(); Console.WriteLine("\n═══ DEMO 1: Direct Agent Conversation ═══\n"); AIAgent biologistProxy = services.GetRequiredKeyedService("Biologist"); -AgentSession session = await biologistProxy.GetNewSessionAsync(); +AgentSession session = await biologistProxy.CreateSessionAsync(); AgentResponse response = await biologistProxy.RunAsync("What is photosynthesis?", session); Console.WriteLine($"🧬 Biologist: {response.Text}\n"); AIAgent chemistProxy = services.GetRequiredKeyedService("Chemist"); -session = await chemistProxy.GetNewSessionAsync(); +session = await chemistProxy.CreateSessionAsync(); response = await chemistProxy.RunAsync("What is a chemical bond?", session); Console.WriteLine($"🧪 Chemist: {response.Text}\n"); diff --git a/dotnet/samples/04-hosting/DurableWorkflows/ConsoleApps/05_WorkflowEvents/Program.cs b/dotnet/samples/04-hosting/DurableWorkflows/ConsoleApps/05_WorkflowEvents/Program.cs index 7ca3ae77c1..3ddec1db37 100644 --- a/dotnet/samples/04-hosting/DurableWorkflows/ConsoleApps/05_WorkflowEvents/Program.cs +++ b/dotnet/samples/04-hosting/DurableWorkflows/ConsoleApps/05_WorkflowEvents/Program.cs @@ -116,7 +116,7 @@ static async Task RunWorkflowWithStreamingAsync(string orderId, Workflow workflo break; case WorkflowOutputEvent e: - WriteColored($" [Output] {e.SourceId}", ConsoleColor.DarkGray); + WriteColored($" [Output] {e.ExecutorId}", ConsoleColor.DarkGray); break; // Workflow completion diff --git a/dotnet/samples/04-hosting/DurableWorkflows/ConsoleApps/06_WorkflowSharedState/Executors.cs b/dotnet/samples/04-hosting/DurableWorkflows/ConsoleApps/06_WorkflowSharedState/Executors.cs index 05afbab71d..c0c8cfd097 100644 --- a/dotnet/samples/04-hosting/DurableWorkflows/ConsoleApps/06_WorkflowSharedState/Executors.cs +++ b/dotnet/samples/04-hosting/DurableWorkflows/ConsoleApps/06_WorkflowSharedState/Executors.cs @@ -1,4 +1,4 @@ -// Copyright (c) Microsoft. All rights reserved. +// Copyright (c) Microsoft. All rights reserved. using Microsoft.Agents.AI.Workflows; diff --git a/dotnet/samples/04-hosting/DurableWorkflows/ConsoleApps/06_WorkflowSharedState/Program.cs b/dotnet/samples/04-hosting/DurableWorkflows/ConsoleApps/06_WorkflowSharedState/Program.cs index 4b46779eb8..2513cc2dad 100644 --- a/dotnet/samples/04-hosting/DurableWorkflows/ConsoleApps/06_WorkflowSharedState/Program.cs +++ b/dotnet/samples/04-hosting/DurableWorkflows/ConsoleApps/06_WorkflowSharedState/Program.cs @@ -93,7 +93,7 @@ while (true) switch (evt) { case WorkflowOutputEvent e: - Console.WriteLine($" [Output] {e.SourceId}: {e.Data}"); + Console.WriteLine($" [Output] {e.ExecutorId}: {e.Data}"); break; case DurableWorkflowCompletedEvent e: diff --git a/dotnet/samples/04-hosting/DurableWorkflows/ConsoleApps/08_WorkflowHITL/Executors.cs b/dotnet/samples/04-hosting/DurableWorkflows/ConsoleApps/08_WorkflowHITL/Executors.cs index db1a873260..2006b1cd19 100644 --- a/dotnet/samples/04-hosting/DurableWorkflows/ConsoleApps/08_WorkflowHITL/Executors.cs +++ b/dotnet/samples/04-hosting/DurableWorkflows/ConsoleApps/08_WorkflowHITL/Executors.cs @@ -1,4 +1,4 @@ -// Copyright (c) Microsoft. All rights reserved. +// Copyright (c) Microsoft. All rights reserved. using Microsoft.Agents.AI.Workflows; diff --git a/dotnet/samples/04-hosting/DurableWorkflows/ConsoleApps/08_WorkflowHITL/Program.cs b/dotnet/samples/04-hosting/DurableWorkflows/ConsoleApps/08_WorkflowHITL/Program.cs index 3b73b2d9b0..bc8fe00341 100644 --- a/dotnet/samples/04-hosting/DurableWorkflows/ConsoleApps/08_WorkflowHITL/Program.cs +++ b/dotnet/samples/04-hosting/DurableWorkflows/ConsoleApps/08_WorkflowHITL/Program.cs @@ -1,4 +1,4 @@ -// Copyright (c) Microsoft. All rights reserved. +// Copyright (c) Microsoft. All rights reserved. // This sample demonstrates a Human-in-the-Loop (HITL) workflow using Durable Tasks. // @@ -43,7 +43,7 @@ Workflow expenseApproval = new WorkflowBuilder(createRequest) .AddEdge(createRequest, managerApproval) .AddEdge(managerApproval, prepareFinanceReview) .AddFanOutEdge(prepareFinanceReview, [budgetApproval, complianceApproval]) - .AddFanInEdge([budgetApproval, complianceApproval], reimburse) + .AddFanInBarrierEdge([budgetApproval, complianceApproval], reimburse) .Build(); IHost host = Host.CreateDefaultBuilder(args) diff --git a/dotnet/tests/Microsoft.Agents.AI.DurableTask.IntegrationTests/ConsoleAppSamplesValidation.cs b/dotnet/tests/Microsoft.Agents.AI.DurableTask.IntegrationTests/ConsoleAppSamplesValidation.cs index b2dcb8292a..8e93ff3b1a 100644 --- a/dotnet/tests/Microsoft.Agents.AI.DurableTask.IntegrationTests/ConsoleAppSamplesValidation.cs +++ b/dotnet/tests/Microsoft.Agents.AI.DurableTask.IntegrationTests/ConsoleAppSamplesValidation.cs @@ -1,4 +1,4 @@ -// Copyright (c) Microsoft. All rights reserved. +// Copyright (c) Microsoft. All rights reserved. using System.Collections.Concurrent; using System.Diagnostics; @@ -581,365 +581,4 @@ public sealed class ConsoleAppSamplesValidation(ITestOutputHelper outputHelper) Assert.True(contentLinesAfterResume > 0, "No content received after resume (expected to continue from cursor, not restart)."); }); } - - private static string GetTargetFramework() - { - string filePath = new Uri(typeof(ConsoleAppSamplesValidation).Assembly.Location).LocalPath; - string directory = Path.GetDirectoryName(filePath)!; - string tfm = Path.GetFileName(directory); - if (tfm.StartsWith("net", StringComparison.OrdinalIgnoreCase)) - { - return tfm; - } - - throw new InvalidOperationException($"Unable to find target framework in path: {filePath}"); - } - - private async Task StartSharedInfrastructureAsync() - { - this._outputHelper.WriteLine("Starting shared infrastructure for console app samples..."); - - // Start DTS emulator - await this.StartDtsEmulatorAsync(); - - // Start Redis - await this.StartRedisAsync(); - - // Wait for infrastructure to be ready - await Task.Delay(TimeSpan.FromSeconds(5)); - } - - private async Task StartDtsEmulatorAsync() - { - // Start DTS emulator if it's not already running - if (!await this.IsDtsEmulatorRunningAsync()) - { - this._outputHelper.WriteLine("Starting DTS emulator..."); - await this.RunCommandAsync("docker", [ - "run", "-d", - "--name", "dts-emulator", - "-p", $"{DtsPort}:8080", - "-e", "DTS_USE_DYNAMIC_TASK_HUBS=true", - "mcr.microsoft.com/dts/dts-emulator:latest" - ]); - } - } - - private async Task StartRedisAsync() - { - if (!await this.IsRedisRunningAsync()) - { - this._outputHelper.WriteLine("Starting Redis..."); - await this.RunCommandAsync("docker", [ - "run", "-d", - "--name", "redis", - "-p", $"{RedisPort}:6379", - "redis:latest" - ]); - } - } - - private async Task IsDtsEmulatorRunningAsync() - { - this._outputHelper.WriteLine($"Checking if DTS emulator is running at http://localhost:{DtsPort}/healthz..."); - - // DTS emulator doesn't support HTTP/1.1, so we need to use HTTP/2.0 - using HttpClient http2Client = new() - { - DefaultRequestVersion = new Version(2, 0), - DefaultVersionPolicy = HttpVersionPolicy.RequestVersionExact - }; - - try - { - using CancellationTokenSource timeoutCts = new(TimeSpan.FromSeconds(30)); - using HttpResponseMessage response = await http2Client.GetAsync(new Uri($"http://localhost:{DtsPort}/healthz"), timeoutCts.Token); - if (response.Content.Headers.ContentLength > 0) - { - string content = await response.Content.ReadAsStringAsync(timeoutCts.Token); - this._outputHelper.WriteLine($"DTS emulator health check response: {content}"); - } - - if (response.IsSuccessStatusCode) - { - this._outputHelper.WriteLine("DTS emulator is running"); - return true; - } - - this._outputHelper.WriteLine($"DTS emulator is not running. Status code: {response.StatusCode}"); - return false; - } - catch (HttpRequestException ex) - { - this._outputHelper.WriteLine($"DTS emulator is not running: {ex.Message}"); - return false; - } - } - - private async Task IsRedisRunningAsync() - { - this._outputHelper.WriteLine($"Checking if Redis is running at localhost:{RedisPort}..."); - - try - { - using CancellationTokenSource timeoutCts = new(TimeSpan.FromSeconds(30)); - ProcessStartInfo startInfo = new() - { - FileName = "docker", - Arguments = "exec redis redis-cli ping", - UseShellExecute = false, - RedirectStandardOutput = true, - RedirectStandardError = true, - CreateNoWindow = true - }; - - using Process process = new() { StartInfo = startInfo }; - if (!process.Start()) - { - this._outputHelper.WriteLine("Failed to start docker exec command"); - return false; - } - - string output = await process.StandardOutput.ReadToEndAsync(timeoutCts.Token); - await process.WaitForExitAsync(timeoutCts.Token); - - if (process.ExitCode == 0 && output.Contains("PONG", StringComparison.OrdinalIgnoreCase)) - { - this._outputHelper.WriteLine("Redis is running"); - return true; - } - - this._outputHelper.WriteLine($"Redis is not running. Exit code: {process.ExitCode}, Output: {output}"); - return false; - } - catch (Exception ex) - { - this._outputHelper.WriteLine($"Redis is not running: {ex.Message}"); - return false; - } - } - - private async Task RunSampleTestAsync(string samplePath, Func, Task> testAction) - { - // Generate a unique TaskHub name for this sample test to prevent cross-test interference - // when multiple tests run together and share the same DTS emulator. - string uniqueTaskHubName = $"sample-{Guid.NewGuid().ToString("N").Substring(0, 6)}"; - - // Start the console app - // Use BlockingCollection to safely read logs asynchronously captured from the process - using BlockingCollection logsContainer = []; - using Process appProcess = this.StartConsoleApp(samplePath, logsContainer, uniqueTaskHubName); - try - { - // Run the test - await testAction(appProcess, logsContainer); - } - catch (OperationCanceledException e) - { - throw new TimeoutException("Core test logic timed out!", e); - } - finally - { - logsContainer.CompleteAdding(); - await this.StopProcessAsync(appProcess); - } - } - - private sealed record OutputLog(DateTime Timestamp, LogLevel Level, string Message); - - /// - /// Writes a line to the process's stdin and flushes it. - /// Logs the input being sent for debugging purposes. - /// - private async Task WriteInputAsync(Process process, string input, CancellationToken cancellationToken) - { - this._outputHelper.WriteLine($"{DateTime.Now:HH:mm:ss.fff} [{process.ProcessName}(in)]: {input}"); - await process.StandardInput.WriteLineAsync(input); - await process.StandardInput.FlushAsync(cancellationToken); - } - - /// - /// Reads a line from the logs queue, filtering for Information level logs (stdout). - /// Returns null if the collection is completed and empty, or if cancellation is requested. - /// - private string? ReadLogLine(BlockingCollection logs, CancellationToken cancellationToken) - { - try - { - while (!cancellationToken.IsCancellationRequested) - { - // Block until a log entry is available or cancellation is requested - // Take will throw OperationCanceledException if cancelled, or InvalidOperationException if collection is completed - OutputLog log = logs.Take(cancellationToken); - - // Check for unhandled exceptions in the logs, which are never expected (but can happen) - if (log.Message.Contains("Unhandled exception")) - { - Assert.Fail("Console app encountered an unhandled exception."); - } - - // Only return Information level logs (stdout), skip Error logs (stderr) - if (log.Level == LogLevel.Information) - { - return log.Message; - } - } - } - catch (OperationCanceledException) - { - // Cancellation requested - return null; - } - catch (InvalidOperationException) - { - // Collection is completed and empty - return null; - } - - return null; - } - - private Process StartConsoleApp(string samplePath, BlockingCollection logs, string taskHubName) - { - ProcessStartInfo startInfo = new() - { - FileName = "dotnet", - Arguments = $"run --framework {s_dotnetTargetFramework}", - WorkingDirectory = samplePath, - UseShellExecute = false, - RedirectStandardOutput = true, - RedirectStandardError = true, - RedirectStandardInput = true, - }; - - string openAiEndpoint = s_configuration["AZURE_OPENAI_ENDPOINT"] ?? - throw new InvalidOperationException("The required AZURE_OPENAI_ENDPOINT env variable is not set."); - string openAiDeployment = s_configuration["AZURE_OPENAI_DEPLOYMENT_NAME"] ?? - throw new InvalidOperationException("The required AZURE_OPENAI_DEPLOYMENT_NAME env variable is not set."); - - void SetAndLogEnvironmentVariable(string key, string value) - { - this._outputHelper.WriteLine($"Setting environment variable for {startInfo.FileName} sub-process: {key}={value}"); - startInfo.EnvironmentVariables[key] = value; - } - - // Set required environment variables for the app - SetAndLogEnvironmentVariable("AZURE_OPENAI_ENDPOINT", openAiEndpoint); - SetAndLogEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT_NAME", openAiDeployment); - SetAndLogEnvironmentVariable("DURABLE_TASK_SCHEDULER_CONNECTION_STRING", - $"Endpoint=http://localhost:{DtsPort};TaskHub={taskHubName};Authentication=None"); - SetAndLogEnvironmentVariable("REDIS_CONNECTION_STRING", $"localhost:{RedisPort}"); - - Process process = new() { StartInfo = startInfo }; - - // Capture the output and error streams asynchronously - // These events fire asynchronously, so we add to the blocking collection which is thread-safe - process.ErrorDataReceived += (sender, e) => - { - if (e.Data != null) - { - string logMessage = $"{DateTime.Now:HH:mm:ss.fff} [{startInfo.FileName}(err)]: {e.Data}"; - this._outputHelper.WriteLine(logMessage); - Debug.WriteLine(logMessage); - try - { - logs.Add(new OutputLog(DateTime.Now, LogLevel.Error, e.Data)); - } - catch (InvalidOperationException) - { - // Collection is completed, ignore - } - } - }; - - process.OutputDataReceived += (sender, e) => - { - if (e.Data != null) - { - string logMessage = $"{DateTime.Now:HH:mm:ss.fff} [{startInfo.FileName}(out)]: {e.Data}"; - this._outputHelper.WriteLine(logMessage); - Debug.WriteLine(logMessage); - try - { - logs.Add(new OutputLog(DateTime.Now, LogLevel.Information, e.Data)); - } - catch (InvalidOperationException) - { - // Collection is completed, ignore - } - } - }; - - if (!process.Start()) - { - throw new InvalidOperationException("Failed to start the console app"); - } - - process.BeginErrorReadLine(); - process.BeginOutputReadLine(); - - return process; - } - - private async Task RunCommandAsync(string command, string[] args) - { - await this.RunCommandAsync(command, workingDirectory: null, args: args); - } - - private async Task RunCommandAsync(string command, string? workingDirectory, string[] args) - { - ProcessStartInfo startInfo = new() - { - FileName = command, - Arguments = string.Join(" ", args), - WorkingDirectory = workingDirectory, - UseShellExecute = false, - RedirectStandardOutput = true, - RedirectStandardError = true, - CreateNoWindow = true - }; - - this._outputHelper.WriteLine($"Running command: {command} {string.Join(" ", args)}"); - - using Process process = new() { StartInfo = startInfo }; - process.ErrorDataReceived += (sender, e) => this._outputHelper.WriteLine($"[{command}(err)]: {e.Data}"); - process.OutputDataReceived += (sender, e) => this._outputHelper.WriteLine($"[{command}(out)]: {e.Data}"); - if (!process.Start()) - { - throw new InvalidOperationException("Failed to start the command"); - } - process.BeginErrorReadLine(); - process.BeginOutputReadLine(); - - using CancellationTokenSource cancellationTokenSource = new(TimeSpan.FromMinutes(1)); - await process.WaitForExitAsync(cancellationTokenSource.Token); - - this._outputHelper.WriteLine($"Command completed with exit code: {process.ExitCode}"); - } - - private async Task StopProcessAsync(Process process) - { - try - { - if (!process.HasExited) - { - this._outputHelper.WriteLine($"{DateTime.Now:HH:mm:ss.fff} Killing process {process.ProcessName}#{process.Id}"); - process.Kill(entireProcessTree: true); - - using CancellationTokenSource timeoutCts = new(TimeSpan.FromSeconds(10)); - await process.WaitForExitAsync(timeoutCts.Token); - this._outputHelper.WriteLine($"{DateTime.Now:HH:mm:ss.fff} Process exited: {process.Id}"); - } - } - catch (Exception ex) - { - this._outputHelper.WriteLine($"{DateTime.Now:HH:mm:ss.fff} Failed to stop process: {ex.Message}"); - } - } - - private CancellationTokenSource CreateTestTimeoutCts(TimeSpan? timeout = null) - { - TimeSpan testTimeout = Debugger.IsAttached ? TimeSpan.FromMinutes(5) : timeout ?? TimeSpan.FromSeconds(60); - return new CancellationTokenSource(testTimeout); - } } diff --git a/dotnet/tests/Microsoft.Agents.AI.DurableTask.UnitTests/Workflows/DurableWorkflowContextTests.cs b/dotnet/tests/Microsoft.Agents.AI.DurableTask.UnitTests/Workflows/DurableWorkflowContextTests.cs index 437f236ee9..7437bb6e1d 100644 --- a/dotnet/tests/Microsoft.Agents.AI.DurableTask.UnitTests/Workflows/DurableWorkflowContextTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.DurableTask.UnitTests/Workflows/DurableWorkflowContextTests.cs @@ -8,7 +8,7 @@ namespace Microsoft.Agents.AI.DurableTask.UnitTests.Workflows; public sealed class DurableWorkflowContextTests { private static FunctionExecutor CreateTestExecutor(string id = "test-executor") - => new(id, (_, _, _) => default); + => new(id, (_, _, _) => default, outputTypes: [typeof(string)]); #region ReadStateAsync