agent/agent-framework
2
0
Fork 0
mirror of https://github.com/microsoft/agent-framework.git synced 2026-06-16 21:04:09 +08:00
Code Issues Packages Projects Releases Wiki Activity

Fix build errors: remove duplicate base class members, update renamed APIs

Browse Source 
- Remove duplicate OutputLog, WriteInputAsync, CreateTestTimeoutCts, etc. from
  ConsoleAppSamplesValidation (already in SamplesValidationBase)
- Update AddFanInEdge -> AddFanInBarrierEdge in workflow samples
- Update GetNewSessionAsync -> CreateSessionAsync in workflow samples
- Update SourceId -> ExecutorId (obsolete) in workflow samples

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
Shyju Krishnankutty
2026-03-03 14:26:47 -08:00
Unverified
parent cd8344dbc1
commit 3c586b25ca
11 changed files with 14 additions and 375 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dotnet/samples/04-hosting/DurableWorkflows/AzureFunctions/02_ConcurrentWorkflow/Program.cs
+1 -1
View File
@@ -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
dotnet/samples/04-hosting/DurableWorkflows/AzureFunctions/03_WorkflowHITL/Program.cs
+1 -1
View File
@@ -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
dotnet/samples/04-hosting/DurableWorkflows/ConsoleApps/02_ConcurrentWorkflow/Program.cs
+1 -1
View File
@@ -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
dotnet/samples/04-hosting/DurableWorkflows/ConsoleApps/04_WorkflowAndAgents/Program.cs
+3 -3
View File
@@ -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<IWorkflowClient>();
Console.WriteLine("\n═══ DEMO 1: Direct Agent Conversation ═══\n");
AIAgent biologistProxy = services.GetRequiredKeyedService<AIAgent>("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<AIAgent>("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");
dotnet/samples/04-hosting/DurableWorkflows/ConsoleApps/05_WorkflowEvents/Program.cs
+1 -1
View File
@@ -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
dotnet/samples/04-hosting/DurableWorkflows/ConsoleApps/06_WorkflowSharedState/Executors.cs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright (c) Microsoft. All rights reserved.
// Copyright (c) Microsoft. All rights reserved.
using Microsoft.Agents.AI.Workflows;
dotnet/samples/04-hosting/DurableWorkflows/ConsoleApps/06_WorkflowSharedState/Program.cs
+1 -1
View File
@@ -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:
dotnet/samples/04-hosting/DurableWorkflows/ConsoleApps/08_WorkflowHITL/Executors.cs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright (c) Microsoft. All rights reserved.
// Copyright (c) Microsoft. All rights reserved.
using Microsoft.Agents.AI.Workflows;
dotnet/samples/04-hosting/DurableWorkflows/ConsoleApps/08_WorkflowHITL/Program.cs
+2 -2
View File
@@ -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)
dotnet/tests/Microsoft.Agents.AI.DurableTask.IntegrationTests/ConsoleAppSamplesValidation.cs
+1 -362
View File
@@ -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<bool> 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<bool> 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<Process, BlockingCollection<OutputLog>, 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<OutputLog> 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);
/// <summary>
/// Writes a line to the process's stdin and flushes it.
/// Logs the input being sent for debugging purposes.
/// </summary>
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);
}
/// <summary>
/// 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.
/// </summary>
private string? ReadLogLine(BlockingCollection<OutputLog> 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<OutputLog> 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);
}
}
dotnet/tests/Microsoft.Agents.AI.DurableTask.UnitTests/Workflows/DurableWorkflowContextTests.cs
+1 -1
View File
@@ -8,7 +8,7 @@ namespace Microsoft.Agents.AI.DurableTask.UnitTests.Workflows;
public sealed class DurableWorkflowContextTests
{
private static FunctionExecutor<string> CreateTestExecutor(string id = "test-executor")
=> new(id, (_, _, _) => default);
=> new(id, (_, _, _) => default, outputTypes: [typeof(string)]);
#region ReadStateAsync