diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Behaviors/BehaviorExecutionException.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Behaviors/BehaviorExecutionException.cs new file mode 100644 index 00000000000..6701a504e05 --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Behaviors/BehaviorExecutionException.cs @@ -0,0 +1,69 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using Microsoft.Shared.Diagnostics; + +namespace Microsoft.Agents.AI.Workflows.Behaviors; + +/// +/// Exception thrown when a behavior fails during execution. +/// +public sealed class BehaviorExecutionException : Exception +{ + /// + /// Initializes a new instance of the class. + /// + public BehaviorExecutionException() + : base("Error executing behavior") + { + this.BehaviorType = string.Empty; + this.Stage = string.Empty; + } + + /// + /// Initializes a new instance of the class with a specified error message. + /// + /// The error message. + public BehaviorExecutionException(string message) + : base(message) + { + this.BehaviorType = string.Empty; + this.Stage = string.Empty; + } + + /// + /// Initializes a new instance of the class with a specified error message and inner exception. + /// + /// The error message. + /// The exception that caused this exception. + public BehaviorExecutionException(string message, Exception innerException) + : base(message, innerException) + { + this.BehaviorType = string.Empty; + this.Stage = string.Empty; + } + + /// + /// Initializes a new instance of the class. + /// + /// The type name of the behavior that failed. + /// The stage at which the behavior failed. + /// The exception that caused the behavior to fail. + public BehaviorExecutionException(string behaviorType, string stage, Exception innerException) + : base($"Error executing behavior '{behaviorType}' at stage '{stage}'", innerException) + { + Throw.IfNull(innerException); + this.BehaviorType = Throw.IfNullOrEmpty(behaviorType); + this.Stage = Throw.IfNullOrEmpty(stage); + } + + /// + /// Gets the type name of the behavior that failed. + /// + public string BehaviorType { get; } + + /// + /// Gets the stage at which the behavior failed. + /// + public string Stage { get; } +} diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Behaviors/BehaviorPipeline.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Behaviors/BehaviorPipeline.cs new file mode 100644 index 00000000000..d6adb206b9e --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Behaviors/BehaviorPipeline.cs @@ -0,0 +1,165 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; + +namespace Microsoft.Agents.AI.Workflows.Behaviors; + +/// +/// Internal class that manages the execution of behavior pipelines for workflows and executors. +/// +internal sealed class BehaviorPipeline +{ + private readonly List _executorBehaviors; + private readonly List _workflowBehaviors; + + /// + /// Initializes a new instance of the class. + /// + /// The collection of executor behaviors to execute. + /// The collection of workflow behaviors to execute. + public BehaviorPipeline( + IEnumerable executorBehaviors, + IEnumerable workflowBehaviors) + { + this._executorBehaviors = executorBehaviors.ToList(); + this._workflowBehaviors = workflowBehaviors.ToList(); + } + + /// + /// Gets a value indicating whether any executor behaviors are registered. + /// + public bool HasExecutorBehaviors => this._executorBehaviors.Count > 0; + + /// + /// Gets a value indicating whether any workflow behaviors are registered. + /// + public bool HasWorkflowBehaviors => this._workflowBehaviors.Count > 0; + + /// + /// Executes the executor behavior pipeline. + /// + /// The context for the executor execution. + /// The final handler to execute after all behaviors. + /// The cancellation token. + /// The result of the executor execution. + public async ValueTask ExecuteExecutorPipelineAsync( + ExecutorBehaviorContext context, + Func> finalHandler, + CancellationToken cancellationToken) + { + if (this._executorBehaviors.Count == 0) + { + return await finalHandler(cancellationToken).ConfigureAwait(false); + } + + // Build chain from end to start (reverse order) + ExecutorBehaviorContinuation pipeline = new(finalHandler); + + for (int i = this._executorBehaviors.Count - 1; i >= 0; i--) + { + var behavior = this._executorBehaviors[i]; + var continuation = pipeline; + pipeline = new ExecutorBehaviorContinuation((ct) => ExecuteBehaviorWithErrorHandlingAsync(behavior, context, continuation, ct)); + } + + return await pipeline(cancellationToken).ConfigureAwait(false); + } + + /// + /// Executes the workflow behavior pipeline with no return value. + /// + /// The context for the workflow execution. + /// The final handler to execute after all behaviors. + /// The cancellation token. + public async ValueTask ExecuteWorkflowPipelineAsync( + WorkflowBehaviorContext context, + Func finalHandler, + CancellationToken cancellationToken) + { + await this.ExecuteWorkflowPipelineAsync( + context, + async ct => { await finalHandler(ct).ConfigureAwait(false); return 0; }, + cancellationToken).ConfigureAwait(false); + } + + /// + /// Executes the workflow behavior pipeline. + /// + /// The result type of the workflow operation. + /// The context for the workflow execution. + /// The final handler to execute after all behaviors. + /// The cancellation token. + /// The result of the workflow execution. + public async ValueTask ExecuteWorkflowPipelineAsync( + WorkflowBehaviorContext context, + Func> finalHandler, + CancellationToken cancellationToken) + { + if (this._workflowBehaviors.Count == 0) + { + return await finalHandler(cancellationToken).ConfigureAwait(false); + } + + // Build chain from end to start (reverse order) + WorkflowBehaviorContinuation pipeline = new(finalHandler); + + for (int i = this._workflowBehaviors.Count - 1; i >= 0; i--) + { + var behavior = this._workflowBehaviors[i]; + var continuation = pipeline; + pipeline = new WorkflowBehaviorContinuation((ct) => ExecuteBehaviorWithErrorHandlingAsync(behavior, context, continuation, ct)); + } + + return await pipeline(cancellationToken).ConfigureAwait(false); + } + + /// + /// Executes an executor behavior with error handling. + /// + private static async ValueTask ExecuteBehaviorWithErrorHandlingAsync( + IExecutorBehavior behavior, + ExecutorBehaviorContext context, + ExecutorBehaviorContinuation continuation, + CancellationToken cancellationToken) + { + try + { + return await behavior.HandleAsync(context, continuation, cancellationToken).ConfigureAwait(false); + } + catch (Exception ex) when (ex is not BehaviorExecutionException) + { + throw new BehaviorExecutionException( + behavior.GetType().FullName ?? "Unknown", + context.Stage.ToString(), + ex + ); + } + } + + /// + /// Executes a workflow behavior with error handling. + /// + private static async ValueTask ExecuteBehaviorWithErrorHandlingAsync( + IWorkflowBehavior behavior, + WorkflowBehaviorContext context, + WorkflowBehaviorContinuation continuation, + CancellationToken cancellationToken) + { + try + { + return await behavior.HandleAsync(context, continuation, cancellationToken).ConfigureAwait(false); + } + catch (Exception ex) when (ex is not BehaviorExecutionException) + { + throw new BehaviorExecutionException( + behavior.GetType().FullName ?? "Unknown", + context.Stage.ToString(), + ex + ); + } + } +} diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Behaviors/IExecutorBehavior.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Behaviors/IExecutorBehavior.cs new file mode 100644 index 00000000000..37e46cd6a65 --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Behaviors/IExecutorBehavior.cs @@ -0,0 +1,114 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; + +namespace Microsoft.Agents.AI.Workflows.Behaviors; + +/// +/// Represents a behavior that wraps executor step execution, allowing custom logic before and after executor operations. +/// +/// +/// Implement this interface to add cross-cutting concerns like logging, telemetry, validation, or performance monitoring +/// at the executor level. Multiple behaviors can be chained together to form a pipeline. +/// Behaviors execute once per executor invocation. Logic placed before await continuation() runs before the executor; +/// logic placed after runs once the executor (and any subsequent behaviors) has completed. +/// +public interface IExecutorBehavior +{ + /// + /// Handles executor execution with the ability to execute logic before and after the next behavior in the pipeline. + /// + /// The context containing information about the current executor execution. + /// The delegate to invoke the next behavior in the pipeline or the actual executor operation. + /// Should be called exactly once. Calling it multiple times will re-execute downstream behaviors and the executor. + /// Logic placed before the call runs before the executor; logic placed after runs once the executor completes. + /// The cancellation token to monitor for cancellation requests. + /// A task representing the asynchronous operation, with the result of the executor operation. + ValueTask HandleAsync( + ExecutorBehaviorContext context, + ExecutorBehaviorContinuation continuation, + CancellationToken cancellationToken = default); +} + +/// +/// Represents the continuation in the executor behavior pipeline. +/// +/// The cancellation token to monitor for cancellation requests. +/// A task representing the asynchronous operation with the executor result. +public delegate ValueTask ExecutorBehaviorContinuation(CancellationToken cancellationToken); + +/// +/// Provides context information for executor behaviors. +/// +public sealed class ExecutorBehaviorContext +{ + /// + /// Gets the string identifier assigned to the executor being invoked. + /// This is the logical name used to register and route messages to the executor, + /// and is distinct from , which represents the executor's CLR type. + /// + public string ExecutorId { get; init; } = string.Empty; + + /// + /// Gets the type of the executor being invoked. + /// + public required Type ExecutorType { get; init; } + + /// + /// Gets the message being processed by the executor. + /// + public required object Message { get; init; } + + /// + /// Gets the type of the message being processed. + /// + public required Type MessageType { get; init; } + + /// + /// Gets the unique identifier for the workflow execution run. + /// + public string RunId { get; init; } = string.Empty; + + /// + /// Gets the stage of executor execution. + /// + public ExecutorStage Stage { get; init; } + + /// + /// Gets the workflow context for this execution. + /// + public required IWorkflowContext WorkflowContext { get; init; } + + /// + /// Gets the trace context for distributed tracing. + /// + public IReadOnlyDictionary? TraceContext { get; init; } + + /// + /// Gets a mutable property bag that behaviors can use to pass additional context to one another. + /// + /// + /// The framework initializes this to an empty dictionary, so behaviors may write to it without a null check. + /// A distinct instance is created for each executor invocation, so values written by an outer behavior are + /// visible to inner behaviors and to code running after await continuation() within the same call, + /// but do not flow to subsequent executor invocations. + /// The dictionary is not thread-safe; behaviors that fan out concurrently must synchronize their own access. + /// + public IDictionary Properties { get; init; } = new Dictionary(); +} + +/// +/// Represents the stage of executor execution. +/// +public enum ExecutorStage +{ + /// + /// Before the executor begins processing the message. Behaviors are invoked once per executor call. + /// To perform logic after the executor completes, place code after the await continuation() call + /// in . + /// + PreExecution +} diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Behaviors/IWorkflowBehavior.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Behaviors/IWorkflowBehavior.cs new file mode 100644 index 00000000000..00003e50ab3 --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Behaviors/IWorkflowBehavior.cs @@ -0,0 +1,103 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; + +namespace Microsoft.Agents.AI.Workflows.Behaviors; + +/// +/// Represents a behavior that wraps workflow execution, allowing custom logic before and after workflow operations. +/// +/// +/// Implement this interface to add cross-cutting concerns like logging, telemetry, validation, or performance monitoring +/// at the workflow level. Multiple behaviors can be chained together to form a pipeline. +/// +public interface IWorkflowBehavior +{ + /// + /// Handles workflow execution with the ability to execute logic before and after the next behavior in the pipeline. + /// + /// The result type of the workflow operation. + /// The context containing information about the current workflow execution. + /// The delegate to invoke the next behavior in the pipeline or the actual workflow operation. + /// Should be called exactly once. Calling it multiple times will re-execute downstream behaviors and the workflow operation. + /// Logic placed before the call runs before the workflow operation; logic placed after runs once the operation completes. + /// The cancellation token to monitor for cancellation requests. + /// A task representing the asynchronous operation, with the result of the workflow operation. + /// + /// Implementations must return the result produced by , or a compatible + /// value. The concrete is determined by the pipeline caller and may vary across invocations; + /// do not assume a specific type or attempt to cast the result to a different type. + /// + ValueTask HandleAsync( + WorkflowBehaviorContext context, + WorkflowBehaviorContinuation continuation, + CancellationToken cancellationToken = default); +} + +/// +/// Represents the continuation in the workflow behavior pipeline. +/// +/// The result type of the operation. +/// The cancellation token to monitor for cancellation requests. +/// A task representing the asynchronous operation. +public delegate ValueTask WorkflowBehaviorContinuation(CancellationToken cancellationToken); + +/// +/// Provides context information for workflow behaviors. +/// +public sealed class WorkflowBehaviorContext +{ + /// + /// Gets the name of the workflow being executed. + /// + public string WorkflowName { get; init; } = string.Empty; + + /// + /// Gets the optional description of the workflow. + /// + public string? WorkflowDescription { get; init; } + + /// + /// Gets the unique identifier for this workflow execution run. + /// + public string RunId { get; init; } = string.Empty; + + /// + /// Gets the identifier of the starting executor in the workflow. + /// + public string StartExecutorId { get; init; } = string.Empty; + + /// + /// Gets the stage of workflow execution. + /// + public WorkflowStage Stage { get; init; } + + /// + /// Gets a mutable property bag that behaviors can use to pass additional context to one another. + /// + /// + /// The framework initializes this to an empty dictionary, so behaviors may write to it without a null check. + /// A distinct instance is created for each invocation, meaning values written + /// during are not visible during . + /// The dictionary is not thread-safe; behaviors that fan out concurrently must synchronize their own access. + /// + public IDictionary Properties { get; init; } = new Dictionary(); +} + +/// +/// Represents the stage of workflow execution. +/// +public enum WorkflowStage +{ + /// + /// The workflow is starting execution. + /// + Starting, + + /// + /// The workflow is ending execution. + /// + Ending +} diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Behaviors/README.md b/dotnet/src/Microsoft.Agents.AI.Workflows/Behaviors/README.md new file mode 100644 index 00000000000..e399b3f0a8d --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Behaviors/README.md @@ -0,0 +1,505 @@ +# Pipeline Behaviors + +Pipeline behaviors provide extension points for adding cross-cutting concerns to workflow and executor execution. They enable developers to inject custom logic before and after workflow operations without modifying core workflow code. + +## Overview + +The pipeline behavior system supports two levels of extensibility: + +- **Workflow Behaviors** - Execute logic when workflows start and end +- **Executor Behaviors** - Execute logic when executors process messages + +Multiple behaviors can be chained together, forming a pipeline where each behavior wraps the next, similar to middleware in ASP.NET Core. + +## Key Features + +- ✅ **Chain of Responsibility Pattern** - Multiple behaviors execute in order +- ✅ **Zero Overhead** - Fast path when no behaviors are registered +- ✅ **Type-Safe Context** - Full access to execution information +- ✅ **Exception Wrapping** - Behaviors wrapped with `BehaviorExecutionException` +- ✅ **Async/Await Support** - Full async execution throughout the pipeline +- ✅ **Short-Circuit Capability** - Behaviors can prevent downstream execution + +## Architecture + +### Execution Flow + +``` +Workflow Start + ├─> WorkflowBehavior 1 (Starting) // entered in registration order + ├─> WorkflowBehavior 2 (Starting) + │ + ├─> SuperStep Loop + │ └─> Executor Message Processing + │ ├─> ExecutorBehavior 1 // code before `await continuation()` + │ ├─> ExecutorBehavior 2 + │ ├─> Actual Handler Execution + │ ├─> ExecutorBehavior 2 // code after `await continuation()` + │ └─> ExecutorBehavior 1 + │ + ├─> WorkflowBehavior 1 (Ending) // a separate pipeline pass, also + └─> WorkflowBehavior 2 (Ending) // entered in registration order +``` + +Executor behaviors form a single nested chain around the handler, so the first-registered behavior is +the outermost one: it runs first on the way in and last on the way out. The `Starting` and `Ending` +workflow stages are two *independent* pipeline passes — `Ending` is not an unwinding of `Starting`, +so both are entered in registration order. + +### Pipeline Behavior Interfaces + +#### IExecutorBehavior + +Wraps individual executor step execution: + +```csharp +public interface IExecutorBehavior +{ + ValueTask HandleAsync( + ExecutorBehaviorContext context, + ExecutorBehaviorContinuation continuation, + CancellationToken cancellationToken = default); +} +``` + +**Context Properties:** +- `ExecutorId` - Unique identifier of the executor +- `ExecutorType` - Type of the executor being invoked +- `Message` - The message being processed +- `MessageType` - Type of the message +- `RunId` - Unique workflow run identifier +- `Stage` - Execution stage. Currently only `PreExecution`: the behavior is invoked once per executor + call, before the executor begins processing the message. To run logic after the executor completes, + place it after the `await continuation()` call rather than looking for a separate post-execution stage. +- `WorkflowContext` - Access to workflow operations +- `TraceContext` - Distributed tracing information +- `Properties` - Mutable `IDictionary`, initialized empty by the framework, for passing + data between behaviors. A fresh instance is created per executor invocation, so values written by an + outer behavior are visible to inner behaviors and to code after `await continuation()` in the same + call, but do not flow to later invocations. Not thread-safe. + +#### IWorkflowBehavior + +Wraps workflow-level execution: + +```csharp +public interface IWorkflowBehavior +{ + ValueTask HandleAsync( + WorkflowBehaviorContext context, + WorkflowBehaviorContinuation continuation, + CancellationToken cancellationToken = default); +} +``` + +**Context Properties:** +- `WorkflowName` - Name of the workflow +- `WorkflowDescription` - Optional description +- `RunId` - Unique workflow run identifier +- `StartExecutorId` - ID of the starting executor +- `Stage` - Workflow execution stage: + - `Starting` - The workflow is beginning execution + - `Ending` - The workflow is completing execution +- `Properties` - Mutable `IDictionary`, initialized empty by the framework, for passing + data between behaviors within a single stage. `Starting` and `Ending` each get their own instance, so + values do not carry across stages. Not thread-safe. + +## Usage Examples + +### Example 1: Logging Behavior + +```csharp +public class LoggingExecutorBehavior : IExecutorBehavior +{ + private readonly ILogger _logger; + + public LoggingExecutorBehavior(ILogger logger) + { + _logger = logger; + } + + public async ValueTask HandleAsync( + ExecutorBehaviorContext context, + ExecutorBehaviorContinuation continuation, + CancellationToken cancellationToken) + { + _logger.LogInformation( + "Executing {ExecutorId} with message type {MessageType}", + context.ExecutorId, + context.MessageType.Name); + + var stopwatch = Stopwatch.StartNew(); + + try + { + var result = await continuation(cancellationToken); + + _logger.LogInformation( + "Completed {ExecutorId} in {ElapsedMs}ms", + context.ExecutorId, + stopwatch.ElapsedMilliseconds); + + return result; + } + catch (Exception ex) + { + _logger.LogError( + ex, + "Failed {ExecutorId} after {ElapsedMs}ms", + context.ExecutorId, + stopwatch.ElapsedMilliseconds); + + throw; + } + } +} +``` + +### Example 2: Validation Behavior + +```csharp +public class ValidationBehavior : IExecutorBehavior +{ + public async ValueTask HandleAsync( + ExecutorBehaviorContext context, + ExecutorBehaviorContinuation continuation, + CancellationToken cancellationToken) + { + // Pre-execution validation + if (context.Message is IValidatable validatable) + { + var validationResult = validatable.Validate(); + if (!validationResult.IsValid) + { + throw new ValidationException( + $"Message validation failed for {context.MessageType.Name}: " + + string.Join(", ", validationResult.Errors)); + } + } + + // Continue pipeline + return await continuation(cancellationToken); + } +} +``` + +### Example 3: Workflow Telemetry Behavior + +```csharp +public class WorkflowTelemetryBehavior : IWorkflowBehavior +{ + private readonly IMetrics _metrics; + + public WorkflowTelemetryBehavior(IMetrics metrics) + { + _metrics = metrics; + } + + public async ValueTask HandleAsync( + WorkflowBehaviorContext context, + WorkflowBehaviorContinuation continuation, + CancellationToken cancellationToken) + { + if (context.Stage == WorkflowStage.Starting) + { + _metrics.IncrementCounter("workflow.starts", 1, + new[] { new KeyValuePair("workflow", context.WorkflowName) }); + } + + var result = await continuation(cancellationToken); + + if (context.Stage == WorkflowStage.Ending) + { + _metrics.IncrementCounter("workflow.completions", 1, + new[] { new KeyValuePair("workflow", context.WorkflowName) }); + } + + return result; + } +} +``` + +### Example 4: Retry Behavior + +```csharp +public class RetryBehavior : IExecutorBehavior +{ + private readonly int _maxRetries; + private readonly TimeSpan _retryDelay; + + public RetryBehavior(int maxRetries = 3, TimeSpan? retryDelay = null) + { + _maxRetries = maxRetries; + _retryDelay = retryDelay ?? TimeSpan.FromMilliseconds(100); + } + + public async ValueTask HandleAsync( + ExecutorBehaviorContext context, + ExecutorBehaviorContinuation continuation, + CancellationToken cancellationToken) + { + for (int attempt = 1; attempt <= _maxRetries; attempt++) + { + try + { + return await continuation(cancellationToken); + } + catch (Exception) when (attempt < _maxRetries) + { + // Wait before retrying + await Task.Delay(_retryDelay, cancellationToken); + } + } + + // Final attempt without catching + return await continuation(cancellationToken); + } +} +``` + +## Registration + +Register behaviors using the `WithBehaviors` method on `WorkflowBuilder`: + +```csharp +var workflow = new WorkflowBuilder(startExecutor) + .WithBehaviors(options => + { + // Add executor behaviors (execute for each executor step) + options.AddExecutorBehavior(new LoggingExecutorBehavior(logger)); + options.AddExecutorBehavior(new ValidationBehavior()); + options.AddExecutorBehavior(new RetryBehavior(maxRetries: 3)); + + // Add workflow behaviors (execute at workflow start/end) + options.AddWorkflowBehavior(new WorkflowTelemetryBehavior(metrics)); + }) + .AddEdge(startExecutor, nextExecutor) + .Build(); +``` + +### Registration with Factory Methods + +For behaviors with parameterless constructors: + +```csharp +.WithBehaviors(options => +{ + options.AddExecutorBehavior(); + options.AddWorkflowBehavior(); +}) +``` + +## Execution Order + +Behaviors execute in the order they are registered: + +```csharp +options.AddExecutorBehavior(new Behavior1()); // Outer wrapper +options.AddExecutorBehavior(new Behavior2()); // Middle wrapper +options.AddExecutorBehavior(new Behavior3()); // Inner wrapper (closest to handler) +``` + +**Execution Flow:** +1. Behavior1.HandleAsync (before) +2. Behavior2.HandleAsync (before) +3. Behavior3.HandleAsync (before) +4. **Actual Handler Execution** +5. Behavior3.HandleAsync (after) +6. Behavior2.HandleAsync (after) +7. Behavior1.HandleAsync (after) + +## Error Handling + +All behavior exceptions are automatically wrapped in `BehaviorExecutionException`: + +```csharp +try +{ + // Execute workflow +} +catch (BehaviorExecutionException ex) +{ + Console.WriteLine($"Behavior {ex.BehaviorType} failed at {ex.Stage}"); + Console.WriteLine($"Inner exception: {ex.InnerException?.Message}"); +} +``` + +**Properties:** +- `BehaviorType` - Full type name of the failed behavior +- `Stage` - Execution stage when failure occurred +- `InnerException` - Original exception thrown by the behavior + +## Performance Considerations + +### Zero Overhead When Disabled + +When no behaviors are registered, the pipeline has zero overhead: + +```csharp +if (_executorBehaviors.Count == 0) +{ + return await finalHandler(cancellationToken); // Direct execution +} +``` + +### Minimal Allocation + +- Behaviors are stored as `List` for optimal iteration +- Delegate chain built once per execution +- No allocations in the fast path + +## Common Use Cases + +### 1. **Logging and Diagnostics** +```csharp +- Log executor execution with timing +- Track message types and payloads +- Record workflow lifecycle events +``` + +### 2. **Telemetry and Metrics** +```csharp +- Count workflow starts/completions +- Measure executor execution time +- Track message processing rates +``` + +### 3. **Validation** +```csharp +- Validate messages before execution +- Enforce business rules +- Check preconditions +``` + +### 4. **Resilience** +```csharp +- Implement retry logic +- Add circuit breakers +- Handle transient failures +``` + +### 5. **Security** +```csharp +- Authorization checks +- Audit logging +- Sensitive data masking +``` + +### 6. **Distributed Tracing** +```csharp +- Create OpenTelemetry spans +- Propagate trace context +- Add custom span attributes +``` + +## Best Practices + +### ✅ Do + +- **Keep behaviors focused** - One concern per behavior +- **Handle cancellation** - Respect `CancellationToken` +- **Use dependency injection** - Pass dependencies via constructor +- **Document side effects** - Be clear about what behaviors do +- **Test behaviors independently** - Unit test each behavior in isolation + +### ❌ Don't + +- **Modify message content** - Behaviors should observe, not mutate +- **Catch all exceptions** - Let exceptions propagate (they'll be wrapped) +- **Block threads** - Always use async/await +- **Share state** - Behaviors should be stateless or thread-safe +- **Assume execution order** - Each behavior should work independently + +## Thread Safety + +Behavior instances may be called concurrently when: +- Multiple workflows execute simultaneously +- The same workflow processes multiple messages in parallel + +**Ensure behaviors are thread-safe:** +- Use immutable state +- Synchronize mutable state access +- Avoid shared static fields + +## Advanced Patterns + +### Conditional Execution + +```csharp +public async ValueTask HandleAsync( + ExecutorBehaviorContext context, + ExecutorBehaviorContinuation continuation, + CancellationToken cancellationToken) +{ + // Only apply to specific executor types + if (context.ExecutorType == typeof(CriticalExecutor)) + { + // Apply special handling + } + + return await continuation(cancellationToken); +} +``` + +### Short-Circuiting + +```csharp +public async ValueTask HandleAsync( + ExecutorBehaviorContext context, + ExecutorBehaviorContinuation continuation, + CancellationToken cancellationToken) +{ + // Check cache + if (_cache.TryGet(context.Message, out var cachedResult)) + { + return cachedResult; // Skip execution + } + + var result = await continuation(cancellationToken); + _cache.Set(context.Message, result); + return result; +} +``` + +### Context Enrichment + +```csharp +public async ValueTask HandleAsync( + ExecutorBehaviorContext context, + ExecutorBehaviorContinuation continuation, + CancellationToken cancellationToken) +{ + // Add custom trace attributes + Activity.Current?.SetTag("executor.id", context.ExecutorId); + Activity.Current?.SetTag("message.type", context.MessageType.Name); + + return await continuation(cancellationToken); +} +``` + +## Backward Compatibility + +Pipeline behaviors are completely opt-in: +- Existing workflows work without modification +- No performance impact when behaviors aren't registered +- New functionality doesn't break existing code + +## API Reference + +### Core Types + +- `IWorkflowBehavior` - Workflow-level behavior interface +- `IExecutorBehavior` - Executor-level behavior interface +- `WorkflowBehaviorContext` - Context for workflow behaviors +- `ExecutorBehaviorContext` - Context for executor behaviors +- `WorkflowBehaviorOptions` - Registration API +- `BehaviorExecutionException` - Exception wrapper + +### Enums + +- `WorkflowStage` - Starting, Ending +- `ExecutorStage` - PreExecution + +### Extension Methods + +- `WorkflowBuilder.WithBehaviors(Action)` - Register behaviors diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Behaviors/WorkflowBehaviorOptions.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Behaviors/WorkflowBehaviorOptions.cs new file mode 100644 index 00000000000..7c204f8a0f9 --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Behaviors/WorkflowBehaviorOptions.cs @@ -0,0 +1,70 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Collections.Generic; +using Microsoft.Shared.Diagnostics; + +namespace Microsoft.Agents.AI.Workflows.Behaviors; + +/// +/// Provides options for configuring workflow and executor behaviors. +/// +public sealed class WorkflowBehaviorOptions +{ + internal List ExecutorBehaviors { get; } = new(); + internal List WorkflowBehaviors { get; } = new(); + + /// + /// Registers an executor behavior instance to the pipeline. + /// + /// The executor behavior instance to register. + /// The current options instance for method chaining. + /// Thrown when is null. + public WorkflowBehaviorOptions AddExecutorBehavior(IExecutorBehavior behavior) + { + this.ExecutorBehaviors.Add(Throw.IfNull(behavior)); + return this; + } + + /// + /// Registers a workflow behavior instance to the pipeline. + /// + /// The workflow behavior instance to register. + /// The current options instance for method chaining. + /// Thrown when is null. + public WorkflowBehaviorOptions AddWorkflowBehavior(IWorkflowBehavior behavior) + { + this.WorkflowBehaviors.Add(Throw.IfNull(behavior)); + return this; + } + + /// + /// Registers an executor behavior using a parameterless constructor. + /// + /// The type of executor behavior to register. + /// The current options instance for method chaining. + public WorkflowBehaviorOptions AddExecutorBehavior() + where TBehavior : IExecutorBehavior, new() + { + return this.AddExecutorBehavior(new TBehavior()); + } + + /// + /// Registers a workflow behavior using a parameterless constructor. + /// + /// The type of workflow behavior to register. + /// The current options instance for method chaining. + public WorkflowBehaviorOptions AddWorkflowBehavior() + where TBehavior : IWorkflowBehavior, new() + { + return this.AddWorkflowBehavior(new TBehavior()); + } + + /// + /// Builds a behavior pipeline from the registered behaviors. + /// + /// A new instance. + internal BehaviorPipeline BuildPipeline() + { + return new BehaviorPipeline(this.ExecutorBehaviors, this.WorkflowBehaviors); + } +} diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Executor.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Executor.cs index 9f092e8e887..e4339ecfa1f 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Executor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Executor.cs @@ -10,6 +10,7 @@ using System.Reflection; using System.Threading; using System.Threading.Tasks; +using Microsoft.Agents.AI.Workflows.Behaviors; using Microsoft.Agents.AI.Workflows.Checkpointing; using Microsoft.Agents.AI.Workflows.Execution; using Microsoft.Agents.AI.Workflows.Observability; @@ -255,6 +256,37 @@ protected internal virtual ValueTask InitializeAsync(IWorkflowContext context, C => this.ExecuteCoreAsync(message, messageType, context, WorkflowTelemetryContext.Disabled, cancellationToken); internal async ValueTask ExecuteCoreAsync(object message, TypeId messageType, IWorkflowContext context, WorkflowTelemetryContext telemetryContext, CancellationToken cancellationToken = default) + => await this.ExecuteCoreAsync(message, messageType, context, telemetryContext, behaviorPipeline: null, runId: null, cancellationToken).ConfigureAwait(false); + + internal async ValueTask ExecuteCoreAsync(object message, TypeId messageType, IWorkflowContext context, WorkflowTelemetryContext telemetryContext, BehaviorPipeline? behaviorPipeline, string? runId, CancellationToken cancellationToken = default) + { + // Check if behaviors are configured + if (behaviorPipeline?.HasExecutorBehaviors == true) + { + var behaviorContext = new ExecutorBehaviorContext + { + ExecutorId = this.Id, + ExecutorType = this.GetType(), + Message = message, + MessageType = message.GetType(), + RunId = runId ?? string.Empty, + Stage = ExecutorStage.PreExecution, + WorkflowContext = context, + TraceContext = context.TraceContext + }; + + return await behaviorPipeline.ExecuteExecutorPipelineAsync( + behaviorContext, + async (ct) => await this.ExecuteCoreInternalAsync(message, messageType, context, telemetryContext, ct).ConfigureAwait(false), + cancellationToken + ).ConfigureAwait(false); + } + + // No behaviors - execute directly (fast path) + return await this.ExecuteCoreInternalAsync(message, messageType, context, telemetryContext, cancellationToken).ConfigureAwait(false); + } + + private async ValueTask ExecuteCoreInternalAsync(object message, TypeId messageType, IWorkflowContext context, WorkflowTelemetryContext telemetryContext, CancellationToken cancellationToken) { using var activity = telemetryContext.StartExecutorProcessActivity(this.Id, this.GetType().FullName, messageType.TypeName, message); activity?.CreateSourceLinks(context.TraceContext); diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunner.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunner.cs index 47167cd9e53..1f9911e26b8 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunner.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunner.cs @@ -7,6 +7,7 @@ using System.Linq; using System.Threading; using System.Threading.Tasks; +using Microsoft.Agents.AI.Workflows.Behaviors; using Microsoft.Agents.AI.Workflows.Checkpointing; using Microsoft.Agents.AI.Workflows.Execution; using Microsoft.Agents.AI.Workflows.Observability; @@ -55,6 +56,7 @@ private InProcessRunner(Workflow workflow, ICheckpointManager? checkpointManager this.Workflow = Throw.IfNull(workflow); this.RunContext = new InProcessRunnerContext(workflow, this.SessionId, checkpointingEnabled: checkpointManager != null, this.OutgoingEvents, this.StepTracer, existingOwnerSignoff, subworkflow, enableConcurrentRuns); + this.RunContext.SetRunEndingCallback(this.ExecuteWorkflowEndBehaviorsAsync); this.CheckpointManager = checkpointManager; this._knownValidInputTypes = knownValidInputTypes != null @@ -160,7 +162,58 @@ private ValueTask RaiseWorkflowEventAsync(WorkflowEvent workflowEvent) public ValueTask BeginStreamAsync(ExecutionMode mode, CancellationToken cancellationToken = default) { this.RunContext.CheckEnded(); - return new(new AsyncRunHandle(this, this, mode)); + + if (this.Workflow.BehaviorPipeline?.HasWorkflowBehaviors != true) + { + return new ValueTask(new AsyncRunHandle(this, this, mode)); + } + + return new ValueTask(this.BeginStreamWithBehaviorsAsync(mode, cancellationToken)); + } + + private async Task BeginStreamWithBehaviorsAsync(ExecutionMode mode, CancellationToken cancellationToken) + { + var context = new WorkflowBehaviorContext + { + WorkflowName = this.Workflow.Name ?? string.Empty, + WorkflowDescription = this.Workflow.Description, + RunId = this.SessionId, + StartExecutorId = this.StartExecutorId, + Stage = WorkflowStage.Starting + }; + + await this.Workflow.BehaviorPipeline!.ExecuteWorkflowPipelineAsync( + context, + (ct) => default, + cancellationToken + ).ConfigureAwait(false); + + return new AsyncRunHandle(this, this, mode); + } + + /// + /// Executes workflow end behaviors if configured. + /// + /// The cancellation token. + internal async ValueTask ExecuteWorkflowEndBehaviorsAsync(CancellationToken cancellationToken = default) + { + if (this.Workflow.BehaviorPipeline?.HasWorkflowBehaviors == true) + { + var context = new WorkflowBehaviorContext + { + WorkflowName = this.Workflow.Name ?? string.Empty, + WorkflowDescription = this.Workflow.Description, + RunId = this.SessionId, + StartExecutorId = this.StartExecutorId, + Stage = WorkflowStage.Ending + }; + + await this.Workflow.BehaviorPipeline.ExecuteWorkflowPipelineAsync( + context, + (ct) => default, + cancellationToken + ).ConfigureAwait(false); + } } public ValueTask ResumeStreamAsync(ExecutionMode mode, CheckpointInfo fromCheckpoint, CancellationToken cancellationToken = default) @@ -264,6 +317,8 @@ await executor.ExecuteCoreAsync( messageType, this.RunContext.BindWorkflowContext(receiverId, envelope.TraceContext), this.TelemetryContext, + this.Workflow.BehaviorPipeline, + this.SessionId, cancellationToken ).ConfigureAwait(false); } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs index 8ccfb957294..684c77d9db1 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs @@ -6,6 +6,7 @@ using System.Diagnostics; using System.Diagnostics.CodeAnalysis; using System.Linq; +using System.Runtime.ExceptionServices; using System.Threading; using System.Threading.Tasks; using Microsoft.Agents.AI.Workflows.Checkpointing; @@ -38,6 +39,8 @@ internal sealed class InProcessRunnerContext : IRunnerContext private readonly ConcurrentDictionary _externalRequests = new(); + private Func? _onRunEnding; + public InProcessRunnerContext( Workflow workflow, string sessionId, @@ -70,6 +73,16 @@ public InProcessRunnerContext( this.ConcurrentRunsEnabled = enableConcurrentRuns; this.OutgoingEvents = outgoingEvents; } + internal void SetRunEndingCallback(Func callback) + { + if (this._onRunEnding is not null) + { + throw new InvalidOperationException("A run-ending callback has already been registered."); + } + + this._onRunEnding = callback; + } + public WorkflowTelemetryContext TelemetryContext => this._workflow.TelemetryContext; public IExternalRequestSink RegisterPort(string executorId, RequestPort port) @@ -500,6 +513,29 @@ public async ValueTask EndRunAsync() { if (Interlocked.Exchange(ref this._runEnded, 1) == 0) { + ExceptionDispatchInfo? runEndingFailure = null; + + if (this._onRunEnding is not null) + { + // CancellationToken.None is intentional here. This call originates from IAsyncDisposable.DisposeAsync() + // (token-less by contract), flows through ISuperStepRunner.RequestEndRunAsync() (also token-less), + // and reaches this point with no token in scope. As a result, behaviors registered for + // WorkflowStage.Ending cannot observe cancellation. A proper fix would require adding a + // CancellationToken overload to ISuperStepRunner.RequestEndRunAsync and threading it through. + try + { + await this._onRunEnding(CancellationToken.None).ConfigureAwait(false); + } + catch (Exception ex) + { + // Teardown must proceed even when an Ending behavior throws. Letting the exception escape here + // would skip executor disposal and workflow ownership release, leaking resources for the rest of + // the process lifetime. The failure is captured and rethrown once teardown completes, so it is + // surfaced to the caller rather than silently swallowed. + runEndingFailure = ExceptionDispatchInfo.Capture(ex); + } + } + foreach (string executorId in this._executors.Keys) { Task executorTask = this._executors[executorId]; @@ -520,6 +556,8 @@ public async ValueTask EndRunAsync() await this._workflow.ReleaseOwnershipAsync(this, this._previousOwnership).ConfigureAwait(false); this._ownsWorkflow = false; } + + runEndingFailure?.Throw(); } } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Microsoft.Agents.AI.Workflows.csproj b/dotnet/src/Microsoft.Agents.AI.Workflows/Microsoft.Agents.AI.Workflows.csproj index 9bd247be911..722626aee3c 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Microsoft.Agents.AI.Workflows.csproj +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Microsoft.Agents.AI.Workflows.csproj @@ -11,6 +11,8 @@ true true true + true + true diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Workflow.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Workflow.cs index 63c94bf8533..3799e4801b9 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Workflow.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Workflow.cs @@ -6,6 +6,7 @@ using System.Linq; using System.Threading; using System.Threading.Tasks; +using Microsoft.Agents.AI.Workflows.Behaviors; using Microsoft.Agents.AI.Workflows.Checkpointing; using Microsoft.Agents.AI.Workflows.Execution; using Microsoft.Agents.AI.Workflows.Observability; @@ -86,6 +87,11 @@ public Dictionary ReflectExecutors() /// internal WorkflowTelemetryContext TelemetryContext { get; } + /// + /// Gets the behavior pipeline for the workflow, if configured. + /// + internal BehaviorPipeline? BehaviorPipeline { get; init; } + internal bool AllowConcurrent => this.ExecutorBindings.Values.All(registration => registration.SupportsConcurrentSharedExecution); internal IEnumerable NonConcurrentExecutorIds => diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowBuilder.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowBuilder.cs index 3acb0c3f612..b0d6959ab6c 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowBuilder.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowBuilder.cs @@ -6,6 +6,7 @@ using System.Linq; using System.Text.Json; using System.Threading; +using Microsoft.Agents.AI.Workflows.Behaviors; using Microsoft.Agents.AI.Workflows.Checkpointing; using Microsoft.Agents.AI.Workflows.Observability; using Microsoft.Shared.Diagnostics; @@ -39,6 +40,7 @@ private readonly record struct EdgeConnection(string SourceId, string TargetId) private string? _name; private string? _description; private WorkflowTelemetryContext _telemetryContext = WorkflowTelemetryContext.Disabled; + private WorkflowBehaviorOptions? _behaviorOptions; /// /// Initializes a new instance of the WorkflowBuilder class with the specified starting executor. @@ -211,6 +213,20 @@ internal void SetTelemetryContext(WorkflowTelemetryContext context) this._telemetryContext = Throw.IfNull(context); } + /// + /// Configures pipeline behaviors for the workflow, allowing custom logic to be executed before and after + /// workflow and executor operations. + /// + /// An action to configure the behavior options. + /// The current instance, enabling fluent configuration. + /// is . + public WorkflowBuilder WithBehaviors(Action configure) + { + this._behaviorOptions ??= new WorkflowBehaviorOptions(); + Throw.IfNull(configure).Invoke(this._behaviorOptions); + return this; + } + /// /// Binds the specified executor (via registration) to the workflow, allowing it to participate in workflow execution. /// @@ -638,7 +654,8 @@ private Workflow BuildInternal(bool validateOrphans, Activity? activity = null) ExecutorBindings = this._executorBindings, Edges = this._edges, Ports = this._requestPorts, - OutputExecutors = this._outputExecutors + OutputExecutors = this._outputExecutors, + BehaviorPipeline = this._behaviorOptions?.BuildPipeline() }; // Using the start executor ID as a proxy for the workflow ID diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Behaviors/BehaviorExecutionExceptionTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Behaviors/BehaviorExecutionExceptionTests.cs new file mode 100644 index 00000000000..5867ad98ade --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Behaviors/BehaviorExecutionExceptionTests.cs @@ -0,0 +1,262 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using FluentAssertions; +using Microsoft.Agents.AI.Workflows.Behaviors; + +namespace Microsoft.Agents.AI.Workflows.UnitTests.Behaviors; + +/// +/// Tests for BehaviorExecutionException error handling and wrapping. +/// +public class BehaviorExecutionExceptionTests +{ + [Fact] + public void Constructor_WithAllParameters_InitializesProperties() + { + // Arrange + const string behaviorType = "TestBehavior"; + const string stage = "PreExecution"; + var innerException = new InvalidOperationException("Inner exception"); + + // Act + var exception = new BehaviorExecutionException(behaviorType, stage, innerException); + + // Assert + exception.BehaviorType.Should().Be(behaviorType); + exception.Stage.Should().Be(stage); + exception.InnerException.Should().Be(innerException); + exception.Message.Should().Contain(behaviorType); + exception.Message.Should().Contain(stage); + } + + [Fact] + public void Constructor_WithNullBehaviorType_ThrowsArgumentNullException() + { + // Arrange + var innerException = new InvalidOperationException(); + + // Act + Action act = () => _ = new BehaviorExecutionException(null!, "stage", innerException); + + // Assert + act.Should().Throw(); + } + + [Fact] + public void Constructor_WithNullStage_ThrowsArgumentNullException() + { + // Arrange + var innerException = new InvalidOperationException(); + + // Act + Action act = () => _ = new BehaviorExecutionException("behavior", null!, innerException); + + // Assert + act.Should().Throw(); + } + + [Fact] + public void Constructor_WithNullInnerException_ThrowsArgumentNullException() + { + // Act + Action act = () => _ = new BehaviorExecutionException("behavior", "stage", null!); + + // Assert + act.Should().Throw(); + } + + [Fact] + public void Message_ContainsBehaviorType() + { + // Arrange + const string behaviorType = "LoggingBehavior"; + var exception = new BehaviorExecutionException( + behaviorType, + "PreExecution", + new InvalidOperationException()); + + // Act + var message = exception.Message; + + // Assert + message.Should().Contain(behaviorType); + } + + [Fact] + public void Message_ContainsStage() + { + // Arrange + const string stage = "Ending"; + var exception = new BehaviorExecutionException( + "TestBehavior", + stage, + new InvalidOperationException()); + + // Act + var message = exception.Message; + + // Assert + message.Should().Contain(stage); + } + + [Fact] + public void Exception_IsSerializable() + { + // Arrange + var exception = new BehaviorExecutionException( + "TestBehavior", + "PreExecution", + new InvalidOperationException("Test")); + + // Act & Assert - Just verify the type is marked as serializable + exception.Should().BeAssignableTo(); + } + + [Fact] + public void InnerException_IsPreserved() + { + // Arrange + const string originalMessage = "Original exception message"; + var innerException = new InvalidOperationException(originalMessage); + + // Act + var exception = new BehaviorExecutionException("TestBehavior", "PreExecution", innerException); + + // Assert + exception.InnerException.Should().NotBeNull(); + exception.InnerException!.Message.Should().Be(originalMessage); + } + + [Fact] + public void StackTrace_IsPreserved() + { + // Arrange + Exception? capturedException = null; + try + { + ThrowTestException(); + } + catch (Exception ex) + { + capturedException = ex; + } + + // Act + var wrappedException = new BehaviorExecutionException( + "TestBehavior", + "PreExecution", + capturedException!); + + // Assert + wrappedException.InnerException!.StackTrace.Should().NotBeNullOrEmpty(); + wrappedException.InnerException!.StackTrace.Should().Contain(nameof(ThrowTestException)); + } + + [Fact] + public void BehaviorType_IsAccessible() + { + // Arrange + const string behaviorType = "MyCustomBehavior"; + var exception = new BehaviorExecutionException( + behaviorType, + "PreExecution", + new InvalidOperationException()); + + // Act + var result = exception.BehaviorType; + + // Assert + result.Should().Be(behaviorType); + } + + [Fact] + public void Stage_IsAccessible() + { + // Arrange + const string stage = "Ending"; + var exception = new BehaviorExecutionException( + "TestBehavior", + stage, + new InvalidOperationException()); + + // Act + var result = exception.Stage; + + // Assert + result.Should().Be(stage); + } + + [Fact] + public void Constructor_Parameterless_UsesDefaultMessageAndEmptyMetadata() + { + // Act + var exception = new BehaviorExecutionException(); + + // Assert + exception.Message.Should().Be("Error executing behavior"); + exception.BehaviorType.Should().BeEmpty(); + exception.Stage.Should().BeEmpty(); + exception.InnerException.Should().BeNull(); + } + + [Fact] + public void Constructor_WithMessage_SetsMessageAndEmptyMetadata() + { + // Act + var exception = new BehaviorExecutionException("custom message"); + + // Assert + exception.Message.Should().Be("custom message"); + exception.BehaviorType.Should().BeEmpty(); + exception.Stage.Should().BeEmpty(); + exception.InnerException.Should().BeNull(); + } + + [Fact] + public void Constructor_WithMessageAndInnerException_SetsBoth() + { + // Arrange + var innerException = new InvalidOperationException("inner"); + + // Act + var exception = new BehaviorExecutionException("custom message", innerException); + + // Assert + exception.Message.Should().Be("custom message"); + exception.InnerException.Should().BeSameAs(innerException); + exception.BehaviorType.Should().BeEmpty(); + exception.Stage.Should().BeEmpty(); + } + + [Fact] + public void Constructor_WithEmptyBehaviorType_ThrowsArgumentException() + { + // Arrange + var innerException = new InvalidOperationException(); + + // Act + Action act = () => _ = new BehaviorExecutionException(string.Empty, "stage", innerException); + + // Assert - empty must be rejected distinctly from null, so the message is never "behavior '' at stage" + act.Should().Throw().Which.Should().NotBeOfType(); + } + + [Fact] + public void Constructor_WithEmptyStage_ThrowsArgumentException() + { + // Arrange + var innerException = new InvalidOperationException(); + + // Act + Action act = () => _ = new BehaviorExecutionException("behavior", string.Empty, innerException); + + // Assert + act.Should().Throw().Which.Should().NotBeOfType(); + } + + private static void ThrowTestException() + { + throw new InvalidOperationException("Test exception"); + } +} diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Behaviors/BehaviorPipelineTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Behaviors/BehaviorPipelineTests.cs new file mode 100644 index 00000000000..b108538bb86 --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Behaviors/BehaviorPipelineTests.cs @@ -0,0 +1,812 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; +using FluentAssertions; +using Microsoft.Agents.AI.Workflows.Behaviors; + +namespace Microsoft.Agents.AI.Workflows.UnitTests.Behaviors; + +public class BehaviorPipelineTests +{ + [Fact] + public async Task ExecutorPipeline_WithNoBehaviors_ReturnsFastPathAsync() + { + // Arrange + var options = new WorkflowBehaviorOptions(); + var pipeline = options.BuildPipeline(); + var executed = false; + + var context = new ExecutorBehaviorContext + { + ExecutorId = "test-executor", + ExecutorType = typeof(BehaviorPipelineTests), + Message = "test", + MessageType = typeof(string), + RunId = Guid.NewGuid().ToString(), + Stage = ExecutorStage.PreExecution, + WorkflowContext = NullWorkflowContext.Instance + }; + + // Act + var result = await pipeline!.ExecuteExecutorPipelineAsync( + context, + async ct => { executed = true; return await Task.FromResult("result"); }, + CancellationToken.None); + + // Assert + executed.Should().BeTrue(); + result.Should().Be("result"); + } + + [Fact] + public async Task ExecutorPipeline_WithNoBehaviors_FinalHandlerExceptionNotWrappedAsync() + { + // Arrange - no behaviors registered, so exceptions from the core handler should not be wrapped + var options = new WorkflowBehaviorOptions(); + var pipeline = options.BuildPipeline(); + + var context = new ExecutorBehaviorContext + { + ExecutorId = "test-executor", + ExecutorType = typeof(BehaviorPipelineTests), + Message = "test", + MessageType = typeof(string), + RunId = Guid.NewGuid().ToString(), + Stage = ExecutorStage.PreExecution, + WorkflowContext = NullWorkflowContext.Instance + }; + + // Act + Func act = async () => await pipeline!.ExecuteExecutorPipelineAsync( + context, + ct => throw new InvalidOperationException("Core handler error"), + CancellationToken.None); + + // Assert - the raw exception propagates without being wrapped in BehaviorExecutionException + await act.Should().ThrowAsync() + .WithMessage("Core handler error"); + } + + [Fact] + public async Task ExecutorPipeline_WithSingleBehavior_ExecutesBehaviorAsync() + { + // Arrange + var behaviorExecuted = false; + var behavior = new TestExecutorBehavior(ctx => behaviorExecuted = true); + + var options = new WorkflowBehaviorOptions(); + options.AddExecutorBehavior(behavior); + var pipeline = options.BuildPipeline(); + + var context = new ExecutorBehaviorContext + { + ExecutorId = "test-executor", + ExecutorType = typeof(BehaviorPipelineTests), + Message = "test", + MessageType = typeof(string), + RunId = Guid.NewGuid().ToString(), + Stage = ExecutorStage.PreExecution, + WorkflowContext = NullWorkflowContext.Instance + }; + + // Act + await pipeline!.ExecuteExecutorPipelineAsync( + context, + async ct => await Task.FromResult("result"), + CancellationToken.None); + + // Assert + behaviorExecuted.Should().BeTrue(); + } + + [Fact] + public async Task ExecutorPipeline_WithMultipleBehaviors_ExecutesInOrderAsync() + { + // Arrange + var executionOrder = new List(); + var behavior1 = new TestExecutorBehavior(ctx => executionOrder.Add(1)); + var behavior2 = new TestExecutorBehavior(ctx => executionOrder.Add(2)); + var behavior3 = new TestExecutorBehavior(ctx => executionOrder.Add(3)); + + var options = new WorkflowBehaviorOptions(); + options.AddExecutorBehavior(behavior1); + options.AddExecutorBehavior(behavior2); + options.AddExecutorBehavior(behavior3); + var pipeline = options.BuildPipeline(); + + var context = new ExecutorBehaviorContext + { + ExecutorId = "test-executor", + ExecutorType = typeof(BehaviorPipelineTests), + Message = "test", + MessageType = typeof(string), + RunId = Guid.NewGuid().ToString(), + Stage = ExecutorStage.PreExecution, + WorkflowContext = NullWorkflowContext.Instance + }; + + // Act + await pipeline!.ExecuteExecutorPipelineAsync( + context, + async ct => await Task.FromResult("result"), + CancellationToken.None); + + // Assert + executionOrder.Should().Equal(1, 2, 3); + } + + [Fact] + public async Task ExecutorPipeline_BehaviorCanShortCircuit_SkipsRemainingPipelineAsync() + { + // Arrange + var behavior1Executed = false; + var behavior2Executed = false; + var coreExecuted = false; + + var behavior1 = new ShortCircuitingExecutorBehavior(() => { behavior1Executed = true; return "short-circuit"; }); + var behavior2 = new TestExecutorBehavior(ctx => behavior2Executed = true); + + var options = new WorkflowBehaviorOptions(); + options.AddExecutorBehavior(behavior1); + options.AddExecutorBehavior(behavior2); + var pipeline = options.BuildPipeline(); + + var context = new ExecutorBehaviorContext + { + ExecutorId = "test-executor", + ExecutorType = typeof(BehaviorPipelineTests), + Message = "test", + MessageType = typeof(string), + RunId = Guid.NewGuid().ToString(), + Stage = ExecutorStage.PreExecution, + WorkflowContext = NullWorkflowContext.Instance + }; + + // Act + var result = await pipeline!.ExecuteExecutorPipelineAsync( + context, + async ct => { coreExecuted = true; return await Task.FromResult("core-result"); }, + CancellationToken.None); + + // Assert + behavior1Executed.Should().BeTrue(); + behavior2Executed.Should().BeFalse(); + coreExecuted.Should().BeFalse(); + result.Should().Be("short-circuit"); + } + + [Fact] + public async Task ExecutorPipeline_BehaviorThrowsException_WrapsInBehaviorExecutionExceptionAsync() + { + // Arrange + var behavior = new ThrowingExecutorBehavior(); + + var options = new WorkflowBehaviorOptions(); + options.AddExecutorBehavior(behavior); + var pipeline = options.BuildPipeline(); + + var context = new ExecutorBehaviorContext + { + ExecutorId = "test-executor", + ExecutorType = typeof(BehaviorPipelineTests), + Message = "test", + MessageType = typeof(string), + RunId = Guid.NewGuid().ToString(), + Stage = ExecutorStage.PreExecution, + WorkflowContext = NullWorkflowContext.Instance + }; + + // Act + Func act = async () => await pipeline!.ExecuteExecutorPipelineAsync( + context, + async ct => await Task.FromResult("result"), + CancellationToken.None); + + // Assert + await act.Should().ThrowAsync() + .WithMessage("*ThrowingExecutorBehavior*"); + } + + [Fact] + public async Task WorkflowPipeline_WithSingleBehavior_ExecutesBehaviorAsync() + { + // Arrange + var behaviorExecuted = false; + var behavior = new TestWorkflowBehavior(ctx => behaviorExecuted = true); + + var options = new WorkflowBehaviorOptions(); + options.AddWorkflowBehavior(behavior); + var pipeline = options.BuildPipeline(); + + var context = new WorkflowBehaviorContext + { + WorkflowName = "test-workflow", + RunId = Guid.NewGuid().ToString(), + StartExecutorId = "start", + Stage = WorkflowStage.Starting + }; + + // Act + await pipeline!.ExecuteWorkflowPipelineAsync( + context, + async ct => await Task.FromResult(0), + CancellationToken.None); + + // Assert + behaviorExecuted.Should().BeTrue(); + } + + [Fact] + public async Task WorkflowPipeline_WithMultipleBehaviors_ExecutesInOrderAsync() + { + // Arrange + var executionOrder = new List(); + var behavior1 = new TestWorkflowBehavior(ctx => executionOrder.Add(1)); + var behavior2 = new TestWorkflowBehavior(ctx => executionOrder.Add(2)); + var behavior3 = new TestWorkflowBehavior(ctx => executionOrder.Add(3)); + + var options = new WorkflowBehaviorOptions(); + options.AddWorkflowBehavior(behavior1); + options.AddWorkflowBehavior(behavior2); + options.AddWorkflowBehavior(behavior3); + var pipeline = options.BuildPipeline(); + + var context = new WorkflowBehaviorContext + { + WorkflowName = "test-workflow", + RunId = Guid.NewGuid().ToString(), + StartExecutorId = "start", + Stage = WorkflowStage.Starting + }; + + // Act + await pipeline!.ExecuteWorkflowPipelineAsync( + context, + async ct => await Task.FromResult(0), + CancellationToken.None); + + // Assert + executionOrder.Should().Equal(1, 2, 3); + } + + [Fact] + public async Task WorkflowPipeline_BehaviorThrowsException_WrapsInBehaviorExecutionExceptionAsync() + { + // Arrange + var behavior = new ThrowingWorkflowBehavior(); + + var options = new WorkflowBehaviorOptions(); + options.AddWorkflowBehavior(behavior); + var pipeline = options.BuildPipeline(); + + var context = new WorkflowBehaviorContext + { + WorkflowName = "test-workflow", + RunId = Guid.NewGuid().ToString(), + StartExecutorId = "start", + Stage = WorkflowStage.Starting + }; + + // Act + Func act = async () => await pipeline!.ExecuteWorkflowPipelineAsync( + context, + async ct => await Task.FromResult(0), + CancellationToken.None); + + // Assert + await act.Should().ThrowAsync() + .WithMessage("*ThrowingWorkflowBehavior*"); + } + + [Fact] + public void HasExecutorBehaviors_WithBehaviors_ReturnsTrue() + { + // Arrange + var options = new WorkflowBehaviorOptions(); + options.AddExecutorBehavior(new TestExecutorBehavior(_ => { })); + var pipeline = options.BuildPipeline(); + + // Act & Assert + pipeline!.HasExecutorBehaviors.Should().BeTrue(); + } + + [Fact] + public void HasExecutorBehaviors_WithoutBehaviors_ReturnsFalse() + { + // Arrange + var options = new WorkflowBehaviorOptions(); + var pipeline = options.BuildPipeline(); + + // Act & Assert + pipeline!.HasExecutorBehaviors.Should().BeFalse(); + } + + [Fact] + public void HasWorkflowBehaviors_WithBehaviors_ReturnsTrue() + { + // Arrange + var options = new WorkflowBehaviorOptions(); + options.AddWorkflowBehavior(new TestWorkflowBehavior(_ => { })); + var pipeline = options.BuildPipeline(); + + // Act & Assert + pipeline!.HasWorkflowBehaviors.Should().BeTrue(); + } + + [Fact] + public void HasWorkflowBehaviors_WithoutBehaviors_ReturnsFalse() + { + // Arrange + var options = new WorkflowBehaviorOptions(); + var pipeline = options.BuildPipeline(); + + // Act & Assert + pipeline!.HasWorkflowBehaviors.Should().BeFalse(); + } + + [Fact] + public async Task WorkflowPipeline_WithNoBehaviors_ReturnsFastPathAsync() + { + // Arrange + var options = new WorkflowBehaviorOptions(); + var pipeline = options.BuildPipeline(); + var executed = false; + + // Act + var result = await pipeline!.ExecuteWorkflowPipelineAsync( + CreateWorkflowContext(), + async ct => { executed = true; return await Task.FromResult(42); }, + CancellationToken.None); + + // Assert + executed.Should().BeTrue(); + result.Should().Be(42); + } + + [Fact] + public async Task WorkflowPipeline_WithNoBehaviors_FinalHandlerExceptionNotWrappedAsync() + { + // Arrange + var options = new WorkflowBehaviorOptions(); + var pipeline = options.BuildPipeline(); + + // Act + Func act = async () => await pipeline!.ExecuteWorkflowPipelineAsync( + CreateWorkflowContext(), + ct => throw new InvalidOperationException("Core handler error"), + CancellationToken.None); + + // Assert - without behaviors the raw exception propagates unwrapped + await act.Should().ThrowAsync().WithMessage("Core handler error"); + } + + [Fact] + public async Task WorkflowPipeline_ReturnsFinalHandlerResultThroughBehaviorsAsync() + { + // Arrange - TResult must survive the trip through the behavior chain + var options = new WorkflowBehaviorOptions(); + options.AddWorkflowBehavior(new TestWorkflowBehavior(_ => { })); + options.AddWorkflowBehavior(new TestWorkflowBehavior(_ => { })); + var pipeline = options.BuildPipeline(); + + // Act + var result = await pipeline!.ExecuteWorkflowPipelineAsync( + CreateWorkflowContext(), + async ct => await Task.FromResult("handler-result"), + CancellationToken.None); + + // Assert + result.Should().Be("handler-result"); + } + + [Fact] + public async Task WorkflowPipeline_BehaviorCanShortCircuit_SkipsRemainingPipelineAsync() + { + // Arrange + var innerRan = false; + var finalHandlerRan = false; + + var options = new WorkflowBehaviorOptions(); + options.AddWorkflowBehavior(new ShortCircuitingWorkflowBehavior("short-circuited")); + options.AddWorkflowBehavior(new TestWorkflowBehavior(_ => innerRan = true)); + var pipeline = options.BuildPipeline(); + + // Act + var result = await pipeline!.ExecuteWorkflowPipelineAsync( + CreateWorkflowContext(), + async ct => { finalHandlerRan = true; return await Task.FromResult("handler-result"); }, + CancellationToken.None); + + // Assert - neither the inner behavior nor the final handler runs + result.Should().Be("short-circuited"); + innerRan.Should().BeFalse(); + finalHandlerRan.Should().BeFalse(); + } + + [Fact] + public async Task WorkflowPipeline_VoidOverload_ExecutesBehaviorsAndFinalHandlerAsync() + { + // Arrange - the non-generic overload is what InProcessRunner uses for Starting/Ending + var executionOrder = new List(); + + var options = new WorkflowBehaviorOptions(); + options.AddWorkflowBehavior(new TestWorkflowBehavior(_ => executionOrder.Add("behavior1"))); + options.AddWorkflowBehavior(new TestWorkflowBehavior(_ => executionOrder.Add("behavior2"))); + var pipeline = options.BuildPipeline(); + + // Act + await pipeline!.ExecuteWorkflowPipelineAsync( + CreateWorkflowContext(), + ct => { executionOrder.Add("final"); return default; }, + CancellationToken.None); + + // Assert + executionOrder.Should().Equal("behavior1", "behavior2", "final"); + } + + [Fact] + public async Task WorkflowPipeline_VoidOverload_WithNoBehaviors_ExecutesFinalHandlerAsync() + { + // Arrange + var options = new WorkflowBehaviorOptions(); + var pipeline = options.BuildPipeline(); + var executed = false; + + // Act + await pipeline!.ExecuteWorkflowPipelineAsync( + CreateWorkflowContext(), + ct => { executed = true; return default; }, + CancellationToken.None); + + // Assert + executed.Should().BeTrue(); + } + + [Fact] + public async Task ExecutorPipeline_BehaviorExecutionException_IsNotDoubleWrappedAsync() + { + // Arrange - an inner behavior's already-wrapped failure must not be re-wrapped by outer behaviors + var options = new WorkflowBehaviorOptions(); + options.AddExecutorBehavior(new TestExecutorBehavior(_ => { })); + options.AddExecutorBehavior(new ThrowingExecutorBehavior()); + var pipeline = options.BuildPipeline(); + + // Act + Func act = async () => await pipeline!.ExecuteExecutorPipelineAsync( + CreateExecutorContext(), + async ct => await Task.FromResult("result"), + CancellationToken.None); + + // Assert - exactly one layer of wrapping, with the original exception underneath + var wrapped = (await act.Should().ThrowAsync()).Which; + wrapped.BehaviorType.Should().Contain(nameof(ThrowingExecutorBehavior)); + wrapped.InnerException.Should().BeOfType(); + } + + [Fact] + public async Task WorkflowPipeline_BehaviorExecutionException_IsNotDoubleWrappedAsync() + { + // Arrange + var options = new WorkflowBehaviorOptions(); + options.AddWorkflowBehavior(new TestWorkflowBehavior(_ => { })); + options.AddWorkflowBehavior(new ThrowingWorkflowBehavior()); + var pipeline = options.BuildPipeline(); + + // Act + Func act = async () => await pipeline!.ExecuteWorkflowPipelineAsync( + CreateWorkflowContext(), + async ct => await Task.FromResult(0), + CancellationToken.None); + + // Assert + var wrapped = (await act.Should().ThrowAsync()).Which; + wrapped.BehaviorType.Should().Contain(nameof(ThrowingWorkflowBehavior)); + wrapped.InnerException.Should().BeOfType(); + } + + [Fact] + public async Task ExecutorPipeline_CancellationTokenIsPropagatedToBehaviorsAndHandlerAsync() + { + // Arrange + using var cts = new CancellationTokenSource(); + CancellationToken observedByBehavior = default; + CancellationToken observedByHandler = default; + + var options = new WorkflowBehaviorOptions(); + options.AddExecutorBehavior(new TokenCapturingExecutorBehavior(ct => observedByBehavior = ct)); + var pipeline = options.BuildPipeline(); + + // Act + await pipeline!.ExecuteExecutorPipelineAsync( + CreateExecutorContext(), + async ct => { observedByHandler = ct; return await Task.FromResult("result"); }, + cts.Token); + + // Assert + observedByBehavior.Should().Be(cts.Token); + observedByHandler.Should().Be(cts.Token); + } + + [Fact] + public async Task WorkflowPipeline_CancellationTokenIsPropagatedToBehaviorsAndHandlerAsync() + { + // Arrange + using var cts = new CancellationTokenSource(); + CancellationToken observedByBehavior = default; + CancellationToken observedByHandler = default; + + var options = new WorkflowBehaviorOptions(); + options.AddWorkflowBehavior(new TokenCapturingWorkflowBehavior(ct => observedByBehavior = ct)); + var pipeline = options.BuildPipeline(); + + // Act + await pipeline!.ExecuteWorkflowPipelineAsync( + CreateWorkflowContext(), + async ct => { observedByHandler = ct; return await Task.FromResult(0); }, + cts.Token); + + // Assert + observedByBehavior.Should().Be(cts.Token); + observedByHandler.Should().Be(cts.Token); + } + + [Fact] + public async Task ExecutorPipeline_BehaviorTransformsResult_ReturnsTransformedValueAsync() + { + // Arrange - behaviors may rewrite the handler's result on the way out + var options = new WorkflowBehaviorOptions(); + options.AddExecutorBehavior(new ResultTransformingExecutorBehavior(r => $"outer({r})")); + options.AddExecutorBehavior(new ResultTransformingExecutorBehavior(r => $"inner({r})")); + var pipeline = options.BuildPipeline(); + + // Act + var result = await pipeline!.ExecuteExecutorPipelineAsync( + CreateExecutorContext(), + async ct => await Task.FromResult("core"), + CancellationToken.None); + + // Assert - the innermost behavior transforms first, the outermost last + result.Should().Be("outer(inner(core))"); + } + + [Fact] + public async Task ExecutorPipeline_MultipleBehaviors_NestInRegistrationOrderAsync() + { + // Arrange - the first registered behavior should be outermost: first in, last out + var log = new List(); + + var options = new WorkflowBehaviorOptions(); + options.AddExecutorBehavior(new NestingExecutorBehavior("A", log)); + options.AddExecutorBehavior(new NestingExecutorBehavior("B", log)); + var pipeline = options.BuildPipeline(); + + // Act + await pipeline!.ExecuteExecutorPipelineAsync( + CreateExecutorContext(), + async ct => { log.Add("handler"); return await Task.FromResult("result"); }, + CancellationToken.None); + + // Assert + log.Should().Equal("A:enter", "B:enter", "handler", "B:exit", "A:exit"); + } + + private static WorkflowBehaviorContext CreateWorkflowContext(WorkflowStage stage = WorkflowStage.Starting) => + new() + { + WorkflowName = "test-workflow", + RunId = Guid.NewGuid().ToString(), + StartExecutorId = "start", + Stage = stage + }; + + private static ExecutorBehaviorContext CreateExecutorContext() => + new() + { + ExecutorId = "test-executor", + ExecutorType = typeof(BehaviorPipelineTests), + Message = "test", + MessageType = typeof(string), + RunId = Guid.NewGuid().ToString(), + Stage = ExecutorStage.PreExecution, + WorkflowContext = NullWorkflowContext.Instance + }; + + // Test helper behaviors + private sealed class TestExecutorBehavior : IExecutorBehavior + { + private readonly Action _action; + + public TestExecutorBehavior(Action action) + { + this._action = action; + } + + public async ValueTask HandleAsync( + ExecutorBehaviorContext context, + ExecutorBehaviorContinuation continuation, + CancellationToken cancellationToken) + { + this._action(context); + return await continuation(cancellationToken); + } + } + + private sealed class ShortCircuitingExecutorBehavior : IExecutorBehavior + { + private readonly Func _resultFactory; + + public ShortCircuitingExecutorBehavior(Func resultFactory) + { + this._resultFactory = resultFactory; + } + + public ValueTask HandleAsync( + ExecutorBehaviorContext context, + ExecutorBehaviorContinuation continuation, + CancellationToken cancellationToken) + { + // Short-circuit: don't call continuation + return new ValueTask(this._resultFactory()); + } + } + + private sealed class ThrowingExecutorBehavior : IExecutorBehavior + { + public ValueTask HandleAsync( + ExecutorBehaviorContext context, + ExecutorBehaviorContinuation continuation, + CancellationToken cancellationToken) + { + throw new InvalidOperationException("Test exception from behavior"); + } + } + + private sealed class TestWorkflowBehavior : IWorkflowBehavior + { + private readonly Action _action; + + public TestWorkflowBehavior(Action action) + { + this._action = action; + } + + public async ValueTask HandleAsync( + WorkflowBehaviorContext context, + WorkflowBehaviorContinuation continuation, + CancellationToken cancellationToken) + { + this._action(context); + return await continuation(cancellationToken); + } + } + + private sealed class ThrowingWorkflowBehavior : IWorkflowBehavior + { + public ValueTask HandleAsync( + WorkflowBehaviorContext context, + WorkflowBehaviorContinuation continuation, + CancellationToken cancellationToken) + { + throw new InvalidOperationException("Test exception from workflow behavior"); + } + } + + private sealed class ShortCircuitingWorkflowBehavior : IWorkflowBehavior + { + private readonly object _result; + + public ShortCircuitingWorkflowBehavior(object result) + { + this._result = result; + } + + public ValueTask HandleAsync( + WorkflowBehaviorContext context, + WorkflowBehaviorContinuation continuation, + CancellationToken cancellationToken) + { + // Short-circuit: don't call continuation + return new ValueTask((TResult)this._result); + } + } + + private sealed class TokenCapturingExecutorBehavior : IExecutorBehavior + { + private readonly Action _capture; + + public TokenCapturingExecutorBehavior(Action capture) + { + this._capture = capture; + } + + public async ValueTask HandleAsync( + ExecutorBehaviorContext context, + ExecutorBehaviorContinuation continuation, + CancellationToken cancellationToken) + { + this._capture(cancellationToken); + return await continuation(cancellationToken); + } + } + + private sealed class TokenCapturingWorkflowBehavior : IWorkflowBehavior + { + private readonly Action _capture; + + public TokenCapturingWorkflowBehavior(Action capture) + { + this._capture = capture; + } + + public async ValueTask HandleAsync( + WorkflowBehaviorContext context, + WorkflowBehaviorContinuation continuation, + CancellationToken cancellationToken) + { + this._capture(cancellationToken); + return await continuation(cancellationToken); + } + } + + private sealed class ResultTransformingExecutorBehavior : IExecutorBehavior + { + private readonly Func _transform; + + public ResultTransformingExecutorBehavior(Func transform) + { + this._transform = transform; + } + + public async ValueTask HandleAsync( + ExecutorBehaviorContext context, + ExecutorBehaviorContinuation continuation, + CancellationToken cancellationToken) + { + var result = await continuation(cancellationToken); + return this._transform(result); + } + } + + private sealed class NestingExecutorBehavior : IExecutorBehavior + { + private readonly string _name; + private readonly List _log; + + public NestingExecutorBehavior(string name, List log) + { + this._name = name; + this._log = log; + } + + public async ValueTask HandleAsync( + ExecutorBehaviorContext context, + ExecutorBehaviorContinuation continuation, + CancellationToken cancellationToken) + { + this._log.Add($"{this._name}:enter"); + var result = await continuation(cancellationToken); + this._log.Add($"{this._name}:exit"); + return result; + } + } + + private sealed class NullWorkflowContext : IWorkflowContext + { + public static readonly NullWorkflowContext Instance = new(); + + public ValueTask AddEventAsync(WorkflowEvent workflowEvent, CancellationToken cancellationToken = default) => default; + public ValueTask SendMessageAsync(object message, string? targetId, CancellationToken cancellationToken = default) => default; + public ValueTask YieldOutputAsync(object output, CancellationToken cancellationToken = default) => default; + public ValueTask RequestHaltAsync() => default; + public ValueTask ReadStateAsync(string key, string? scopeName = null, CancellationToken cancellationToken = default) => default; + public ValueTask ReadOrInitStateAsync(string key, Func initialStateFactory, string? scopeName = null, CancellationToken cancellationToken = default) => new(initialStateFactory()); + public ValueTask> ReadStateKeysAsync(string? scopeName = null, CancellationToken cancellationToken = default) => new(new HashSet()); + public ValueTask QueueStateUpdateAsync(string key, T? value, string? scopeName = null, CancellationToken cancellationToken = default) => default; + public ValueTask QueueClearScopeAsync(string? scopeName = null, CancellationToken cancellationToken = default) => default; + public IReadOnlyDictionary? TraceContext => null; + public bool ConcurrentRunsEnabled => false; + } +} diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Behaviors/WorkflowBehaviorEndToEndTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Behaviors/WorkflowBehaviorEndToEndTests.cs new file mode 100644 index 00000000000..e9bedf85e80 --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Behaviors/WorkflowBehaviorEndToEndTests.cs @@ -0,0 +1,820 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Collections.Generic; +using System.Diagnostics; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using FluentAssertions; +using Microsoft.Agents.AI.Workflows.Behaviors; +using Microsoft.Agents.AI.Workflows.Execution; +using Microsoft.Agents.AI.Workflows.InProc; + +namespace Microsoft.Agents.AI.Workflows.UnitTests.Behaviors; + +/// +/// End-to-end tests that validate pipeline behaviors work with actual workflows. +/// +public class WorkflowBehaviorEndToEndTests +{ + [Fact] + public async Task Workflow_WithExecutorBehavior_BehaviorExecutesBeforeAndAfterExecutorAsync() + { + // Arrange + var executionLog = new List(); + var behavior = new LoggingExecutorBehavior(executionLog); + + var executor1 = new LoggingExecutor("executor1", executionLog); + var executor2 = new LoggingExecutor("executor2", executionLog); + + var workflow = new WorkflowBuilder(executor1) + .WithBehaviors(options => options.AddExecutorBehavior(behavior)) + .AddEdge(executor1, executor2) + .Build(); + + // Act + await using var run = await InProcessExecution.RunAsync(workflow, "test-input"); + + // Assert + executionLog.Should().ContainInOrder( + "Behavior:PreExecution:executor1", + "Executor:executor1", + "Behavior:PreExecution:executor2", + "Executor:executor2" + ); + } + + [Fact] + public async Task Workflow_WithWorkflowBehavior_BehaviorExecutesAtStartAsync() + { + // Arrange + var executionLog = new List(); + var behavior = new LoggingWorkflowBehavior(executionLog); + + var executor = new LoggingExecutor("executor", executionLog); + + var workflow = new WorkflowBuilder(executor) + .WithBehaviors(options => options.AddWorkflowBehavior(behavior)) + .Build(); + + // Act + await using var run = await InProcessExecution.RunAsync(workflow, "test-input"); + + // Assert - workflow start behavior should execute before executor + executionLog.Should().Contain("WorkflowBehavior:Starting"); + executionLog.Should().Contain("Executor:executor"); + + var startIndex = executionLog.IndexOf("WorkflowBehavior:Starting"); + var executorIndex = executionLog.IndexOf("Executor:executor"); + startIndex.Should().BeLessThan(executorIndex); + } + + [Fact] + public async Task Workflow_WithWorkflowBehavior_BehaviorExecutesAtEndAsync() + { + // Arrange + var executionLog = new List(); + var behavior = new LoggingWorkflowBehavior(executionLog); + var executor = new SimpleExecutor("executor"); + + var workflow = new WorkflowBuilder(executor) + .WithBehaviors(options => options.AddWorkflowBehavior(behavior)) + .Build(); + + // Act - dispose triggers the Ending stage + await using (await InProcessExecution.RunAsync(workflow, "test-input")) + { + } + + // Assert - both Starting and Ending stages should execute, in order + executionLog.Should().Contain("WorkflowBehavior:Starting"); + executionLog.Should().Contain("WorkflowBehavior:Ending"); + + var startIndex = executionLog.IndexOf("WorkflowBehavior:Starting"); + var endIndex = executionLog.IndexOf("WorkflowBehavior:Ending"); + startIndex.Should().BeLessThan(endIndex); + } + + [Fact] + public async Task Workflow_WithBothBehaviorTypes_ExecutesInCorrectOrderAsync() + { + // Arrange + var executionLog = new List(); + var workflowBehavior = new LoggingWorkflowBehavior(executionLog); + var executorBehavior = new LoggingExecutorBehavior(executionLog); + var executor = new LoggingExecutor("executor", executionLog); + + var workflow = new WorkflowBuilder(executor) + .WithBehaviors(options => + { + options.AddWorkflowBehavior(workflowBehavior); + options.AddExecutorBehavior(executorBehavior); + }) + .Build(); + + // Act + await using (await InProcessExecution.RunAsync(workflow, "test-input")) + { + } + + // Assert - Starting → PreExecution (with executor) → Ending + executionLog.Should().ContainInOrder( + "WorkflowBehavior:Starting", + "Behavior:PreExecution:executor", + "Executor:executor", + "WorkflowBehavior:Ending" + ); + } + + [Fact] + public async Task Workflow_WithBehaviors_RunIdIsConsistentAcrossContextsAsync() + { + // Arrange + string? workflowBehaviorRunId = null; + string? executorBehaviorRunId = null; + + var workflowBehavior = new CapturingWorkflowBehavior(ctx => workflowBehaviorRunId = ctx.RunId); + var executorBehavior = new CapturingExecutorBehavior(ctx => executorBehaviorRunId = ctx.RunId); + + var executor = new SimpleExecutor("executor"); + var workflow = new WorkflowBuilder(executor) + .WithBehaviors(options => + { + options.AddWorkflowBehavior(workflowBehavior); + options.AddExecutorBehavior(executorBehavior); + }) + .Build(); + + // Act + string runId; + await using (var run = await InProcessExecution.RunAsync(workflow, "test-input")) + { + runId = run.SessionId; + } + + // Assert - all behavior contexts share the same RunId as the run itself + workflowBehaviorRunId.Should().NotBeNullOrEmpty(); + executorBehaviorRunId.Should().NotBeNullOrEmpty(); + workflowBehaviorRunId.Should().Be(runId); + executorBehaviorRunId.Should().Be(runId); + } + + [Fact] + public async Task Workflow_WithMultipleBehaviors_AllBehaviorsExecuteAsync() + { + // Arrange + var executionLog = new List(); + var loggingBehavior = new LoggingExecutorBehavior(executionLog); + var validationBehavior = new ValidationExecutorBehavior(executionLog); + + var executor = new LoggingExecutor("executor", executionLog); + + var workflow = new WorkflowBuilder(executor) + .WithBehaviors(options => + { + options.AddExecutorBehavior(loggingBehavior); + options.AddExecutorBehavior(validationBehavior); + }) + .Build(); + + // Act + await using var run = await InProcessExecution.RunAsync(workflow, "test-input"); + + // Assert - both behaviors should execute + executionLog.Should().Contain("Behavior:PreExecution:executor"); + executionLog.Should().Contain("Validation:PreExecution:executor"); + executionLog.Should().Contain("Executor:executor"); + } + + [Fact] + public async Task Workflow_WithPerformanceMonitoringBehavior_MeasuresExecutionTimeAsync() + { + // Arrange + var measurements = new Dictionary(); + var behavior = new PerformanceMonitoringBehavior(measurements); + + var executor = new DelayExecutor("slow-executor", TimeSpan.FromMilliseconds(50)); + + var workflow = new WorkflowBuilder(executor) + .WithBehaviors(options => options.AddExecutorBehavior(behavior)) + .Build(); + + // Act + await using var run = await InProcessExecution.RunAsync(workflow, "test-input"); + + // Assert + measurements.Should().ContainKey("slow-executor"); + measurements["slow-executor"].Should().BeGreaterThanOrEqualTo(50); + } + + [Fact] + public async Task Workflow_WithoutBehaviors_ExecutesNormallyAsync() + { + // Arrange + var executionLog = new List(); + var executor = new LoggingExecutor("executor", executionLog); + + var workflow = new WorkflowBuilder(executor).Build(); + + // Act + await using var run = await InProcessExecution.RunAsync(workflow, "test-input"); + + // Assert - workflow executes normally without behaviors + executionLog.Should().Contain("Executor:executor"); + } + + [Fact] + public async Task Workflow_BehaviorShortCircuits_ExecutorDoesNotRunAsync() + { + // Arrange + var executionLog = new List(); + var shortCircuitBehavior = new ShortCircuitBehavior("short-circuit-result"); + + var executor = new LoggingExecutor("executor", executionLog); + + var workflow = new WorkflowBuilder(executor) + .WithBehaviors(options => options.AddExecutorBehavior(shortCircuitBehavior)) + .Build(); + + // Act + await using var run = await InProcessExecution.RunAsync(workflow, "test-input"); + + // Assert - executor should not execute due to short-circuit + executionLog.Should().BeEmpty(); + } + + [Fact] + public async Task Workflow_BehaviorThrowsException_EmitsErrorEventAsync() + { + // Arrange + var faultyBehavior = new FaultyBehavior(); + var executor = new SimpleExecutor("executor"); + + var workflow = new WorkflowBuilder(executor) + .WithBehaviors(options => options.AddExecutorBehavior(faultyBehavior)) + .Build(); + + // Act - exceptions from behaviors are caught and emitted as WorkflowErrorEvent, not thrown + await using var run = await InProcessExecution.RunAsync(workflow, "test-input"); + + // Assert + var behaviorException = run.OutgoingEvents.OfType() + .Should().ContainSingle().Which.Exception + .Should().BeOfType().Subject; + + behaviorException.BehaviorType.Should().Contain(nameof(FaultyBehavior)); + behaviorException.Stage.Should().Be(nameof(ExecutorStage.PreExecution)); + } + + [Fact] + public async Task Workflow_WithExecutorBehaviors_PropertiesFlowBetweenBehaviorsAsync() + { + // Arrange - an outer behavior enriches the context, an inner behavior reads what it wrote + object? observedByInner = null; + var enriching = new EnrichingExecutorBehavior("tenant-id", "contoso"); + var reading = new CapturingExecutorBehavior( + ctx => observedByInner = ctx.Properties.TryGetValue("tenant-id", out var value) ? value : null); + + var executor = new SimpleExecutor("executor"); + var workflow = new WorkflowBuilder(executor) + .WithBehaviors(options => + { + options.AddExecutorBehavior(enriching); + options.AddExecutorBehavior(reading); + }) + .Build(); + + // Act + await using var run = await InProcessExecution.RunAsync(workflow, "test-input"); + + // Assert - the property bag is framework-initialized and shared down the chain + observedByInner.Should().Be("contoso"); + } + + [Fact] + public async Task Workflow_EndingBehaviorThrows_ExecutorsAreStillDisposedAsync() + { + // Arrange + var disposedExecutors = new List(); + var faultyBehavior = new FaultyEndingWorkflowBehavior(); + var asyncExecutor = new DisposableExecutor("async-disposable", disposedExecutors); + var syncExecutor = new SyncDisposableExecutor("sync-disposable", disposedExecutors); + + var workflow = new WorkflowBuilder(asyncExecutor) + .WithBehaviors(options => options.AddWorkflowBehavior(faultyBehavior)) + .AddEdge(asyncExecutor, syncExecutor) + .Build(); + + var run = await InProcessExecution.RunAsync(workflow, "test-input"); + + // Act - the Ending stage runs during disposal + Func disposeRun = async () => await run.DisposeAsync(); + + // Assert - the failure is surfaced to the caller... + await disposeRun.Should().ThrowAsync(); + + // ...but teardown still ran to completion, so executors were disposed rather than leaked. + disposedExecutors.Should().Contain("async-disposable"); + disposedExecutors.Should().Contain("sync-disposable"); + } + + [Fact] + public async Task Workflow_ExecutorBehaviorContext_IsFullyPopulatedAsync() + { + // Arrange + ExecutorBehaviorContext? captured = null; + var behavior = new CapturingExecutorBehavior(ctx => captured ??= ctx); + + var executor = new SimpleExecutor("the-executor"); + var workflow = new WorkflowBuilder(executor) + .WithBehaviors(options => options.AddExecutorBehavior(behavior)) + .Build(); + + // Act + await using var run = await InProcessExecution.RunAsync(workflow, "test-input"); + + // Assert - every context field the public API advertises is populated + captured.Should().NotBeNull(); + captured!.ExecutorId.Should().Be("the-executor"); + captured.ExecutorType.Should().Be(); + captured.Message.Should().Be("test-input"); + captured.MessageType.Should().Be(); + captured.Stage.Should().Be(ExecutorStage.PreExecution); + captured.WorkflowContext.Should().NotBeNull(); + captured.RunId.Should().NotBeNullOrEmpty(); + captured.Properties.Should().NotBeNull(); + } + + [Fact] + public async Task Workflow_WorkflowBehaviorContext_IsFullyPopulatedAsync() + { + // Arrange + WorkflowBehaviorContext? captured = null; + var behavior = new CapturingWorkflowBehavior(ctx => captured ??= ctx); + + var executor = new SimpleExecutor("start-executor"); + var workflow = new WorkflowBuilder(executor) + .WithName("my-workflow") + .WithDescription("my-description") + .WithBehaviors(options => options.AddWorkflowBehavior(behavior)) + .Build(); + + // Act + await using var run = await InProcessExecution.RunAsync(workflow, "test-input"); + + // Assert + captured.Should().NotBeNull(); + captured!.WorkflowName.Should().Be("my-workflow"); + captured.WorkflowDescription.Should().Be("my-description"); + captured.StartExecutorId.Should().Be("start-executor"); + captured.Stage.Should().Be(WorkflowStage.Starting); + captured.RunId.Should().NotBeNullOrEmpty(); + captured.Properties.Should().NotBeNull(); + } + + [Fact] + public async Task Workflow_WorkflowBehaviorProperties_AreIsolatedBetweenStagesAsync() + { + // Arrange - Starting and Ending are separate pipeline passes with separate contexts + bool? endingSawStartingValue = null; + var behavior = new StagePropertyBehavior(saw => endingSawStartingValue = saw); + + var executor = new SimpleExecutor("executor"); + var workflow = new WorkflowBuilder(executor) + .WithBehaviors(options => options.AddWorkflowBehavior(behavior)) + .Build(); + + // Act - disposal triggers the Ending stage + await using (await InProcessExecution.RunAsync(workflow, "test-input")) + { + } + + // Assert - values written during Starting do not leak into Ending + endingSawStartingValue.Should().BeFalse(); + } + + [Fact] + public async Task Workflow_ExecutorBehaviorProperties_AreIsolatedBetweenInvocationsAsync() + { + // Arrange - each executor invocation gets a fresh property bag + var sawExistingValue = new List(); + var behavior = new PerInvocationPropertyBehavior(sawExistingValue); + + var executor1 = new SimpleExecutor("executor1"); + var executor2 = new SimpleExecutor("executor2"); + var workflow = new WorkflowBuilder(executor1) + .WithBehaviors(options => options.AddExecutorBehavior(behavior)) + .AddEdge(executor1, executor2) + .Build(); + + // Act + await using var run = await InProcessExecution.RunAsync(workflow, "test-input"); + + // Assert - two invocations, neither of which observed the other's value + sawExistingValue.Should().HaveCountGreaterThanOrEqualTo(2); + sawExistingValue.Should().AllSatisfy(saw => saw.Should().BeFalse()); + } + + [Fact] + public async Task Workflow_EndingBehaviors_RunInRegistrationOrderAsync() + { + // Arrange - Ending is a separate pass, not an unwinding of Starting + var executionLog = new List(); + var executor = new SimpleExecutor("executor"); + + var workflow = new WorkflowBuilder(executor) + .WithBehaviors(options => + { + options.AddWorkflowBehavior(new NamedWorkflowBehavior("first", executionLog)); + options.AddWorkflowBehavior(new NamedWorkflowBehavior("second", executionLog)); + }) + .Build(); + + // Act + await using (await InProcessExecution.RunAsync(workflow, "test-input")) + { + } + + // Assert - both stages are entered in registration order + executionLog.Should().Equal( + "first:Starting", + "second:Starting", + "first:Ending", + "second:Ending"); + } + + [Fact] + public void RunnerContext_SetRunEndingCallback_CalledTwice_Throws() + { + // Arrange - the runner registers exactly one Ending callback; a second must not silently overwrite it + var workflow = new WorkflowBuilder(new SimpleExecutor("executor")).Build(); + var context = new InProcessRunnerContext( + workflow, + sessionId: "test-session", + checkpointingEnabled: false, + outgoingEvents: new ConcurrentEventSink(), + stepTracer: null); + + context.SetRunEndingCallback(_ => default); + + // Act + Action act = () => context.SetRunEndingCallback(_ => default); + + // Assert + act.Should().Throw() + .WithMessage("*already been registered*"); + } + + // Test Executors + private sealed class LoggingExecutor : Executor + { + private readonly List _log; + + public LoggingExecutor(string id, List log) : base(id) + { + this._log = log; + } + + protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder) => + protocolBuilder.ConfigureRoutes(routeBuilder => routeBuilder.AddHandler(async (message, context, ct) => + { + this._log.Add($"Executor:{this.Id}"); + await context.SendMessageAsync(message, ct); + return message; + })); + } + + private sealed class SimpleExecutor : Executor + { + public SimpleExecutor(string id) : base(id) { } + + protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder) => + protocolBuilder.ConfigureRoutes(routeBuilder => routeBuilder.AddHandler(async (message, context, ct) => + { + await context.SendMessageAsync(message, ct); + return message; + })); + } + + private sealed class DisposableExecutor : Executor, IAsyncDisposable + { + private readonly List _disposed; + + public DisposableExecutor(string id, List disposed) : base(id) + { + this._disposed = disposed; + } + + protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder) => + protocolBuilder.ConfigureRoutes(routeBuilder => routeBuilder.AddHandler(async (message, context, ct) => + { + await context.SendMessageAsync(message, ct); + return message; + })); + + public ValueTask DisposeAsync() + { + this._disposed.Add(this.Id); + return default; + } + } + + private sealed class SyncDisposableExecutor : Executor, IDisposable + { + private readonly List _disposed; + + public SyncDisposableExecutor(string id, List disposed) : base(id) + { + this._disposed = disposed; + } + + protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder) => + protocolBuilder.ConfigureRoutes(routeBuilder => routeBuilder.AddHandler(async (message, context, ct) => + { + await context.SendMessageAsync(message, ct); + return message; + })); + + public void Dispose() => this._disposed.Add(this.Id); + } + + private sealed class DelayExecutor : Executor + { + private readonly TimeSpan _delay; + + public DelayExecutor(string id, TimeSpan delay) : base(id) + { + this._delay = delay; + } + + protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder) => + protocolBuilder.ConfigureRoutes(routeBuilder => routeBuilder.AddHandler(async (message, context, ct) => + { + await Task.Delay(this._delay, ct); + await context.SendMessageAsync(message, ct); + return message; + })); + } + + // Test Behaviors + private sealed class LoggingExecutorBehavior : IExecutorBehavior + { + private readonly List _log; + + public LoggingExecutorBehavior(List log) + { + this._log = log; + } + + public async ValueTask HandleAsync( + ExecutorBehaviorContext context, + ExecutorBehaviorContinuation continuation, + CancellationToken cancellationToken) + { + this._log.Add($"Behavior:{context.Stage}:{context.ExecutorId}"); + return await continuation(cancellationToken); + } + } + + private sealed class LoggingWorkflowBehavior : IWorkflowBehavior + { + private readonly List _log; + + public LoggingWorkflowBehavior(List log) + { + this._log = log; + } + + public async ValueTask HandleAsync( + WorkflowBehaviorContext context, + WorkflowBehaviorContinuation continuation, + CancellationToken cancellationToken) + { + this._log.Add($"WorkflowBehavior:{context.Stage}"); + return await continuation(cancellationToken); + } + } + + private sealed class EnrichingExecutorBehavior : IExecutorBehavior + { + private readonly string _key; + private readonly object _value; + + public EnrichingExecutorBehavior(string key, object value) + { + this._key = key; + this._value = value; + } + + public async ValueTask HandleAsync( + ExecutorBehaviorContext context, + ExecutorBehaviorContinuation continuation, + CancellationToken cancellationToken) + { + context.Properties[this._key] = this._value; + return await continuation(cancellationToken); + } + } + + private sealed class StagePropertyBehavior : IWorkflowBehavior + { + private readonly Action _reportEndingSawValue; + + public StagePropertyBehavior(Action reportEndingSawValue) + { + this._reportEndingSawValue = reportEndingSawValue; + } + + public async ValueTask HandleAsync( + WorkflowBehaviorContext context, + WorkflowBehaviorContinuation continuation, + CancellationToken cancellationToken) + { + if (context.Stage == WorkflowStage.Starting) + { + context.Properties["from-starting"] = "value"; + } + else + { + this._reportEndingSawValue(context.Properties.ContainsKey("from-starting")); + } + + return await continuation(cancellationToken); + } + } + + private sealed class PerInvocationPropertyBehavior : IExecutorBehavior + { + private readonly List _sawExistingValue; + + public PerInvocationPropertyBehavior(List sawExistingValue) + { + this._sawExistingValue = sawExistingValue; + } + + public async ValueTask HandleAsync( + ExecutorBehaviorContext context, + ExecutorBehaviorContinuation continuation, + CancellationToken cancellationToken) + { + this._sawExistingValue.Add(context.Properties.ContainsKey("marker")); + context.Properties["marker"] = "set"; + return await continuation(cancellationToken); + } + } + + private sealed class NamedWorkflowBehavior : IWorkflowBehavior + { + private readonly string _name; + private readonly List _log; + + public NamedWorkflowBehavior(string name, List log) + { + this._name = name; + this._log = log; + } + + public async ValueTask HandleAsync( + WorkflowBehaviorContext context, + WorkflowBehaviorContinuation continuation, + CancellationToken cancellationToken) + { + this._log.Add($"{this._name}:{context.Stage}"); + return await continuation(cancellationToken); + } + } + + private sealed class FaultyEndingWorkflowBehavior : IWorkflowBehavior + { + public async ValueTask HandleAsync( + WorkflowBehaviorContext context, + WorkflowBehaviorContinuation continuation, + CancellationToken cancellationToken) + { + if (context.Stage == WorkflowStage.Ending) + { + throw new InvalidOperationException("Ending behavior failed"); + } + + return await continuation(cancellationToken); + } + } + + private sealed class ValidationExecutorBehavior : IExecutorBehavior + { + private readonly List _log; + + public ValidationExecutorBehavior(List log) + { + this._log = log; + } + + public async ValueTask HandleAsync( + ExecutorBehaviorContext context, + ExecutorBehaviorContinuation continuation, + CancellationToken cancellationToken) + { + this._log.Add($"Validation:{context.Stage}:{context.ExecutorId}"); + if (context.Message == null) + { + throw new InvalidOperationException("Message cannot be null"); + } + return await continuation(cancellationToken); + } + } + + private sealed class PerformanceMonitoringBehavior : IExecutorBehavior + { + private readonly Dictionary _measurements; + + public PerformanceMonitoringBehavior(Dictionary measurements) + { + this._measurements = measurements; + } + + public async ValueTask HandleAsync( + ExecutorBehaviorContext context, + ExecutorBehaviorContinuation continuation, + CancellationToken cancellationToken) + { + if (context.Stage == ExecutorStage.PreExecution) + { + var stopwatch = Stopwatch.StartNew(); + var result = await continuation(cancellationToken); + stopwatch.Stop(); + this._measurements[context.ExecutorId] = stopwatch.ElapsedMilliseconds; + return result; + } + + return await continuation(cancellationToken); + } + } + + private sealed class ShortCircuitBehavior : IExecutorBehavior + { + private readonly object _result; + + public ShortCircuitBehavior(object result) + { + this._result = result; + } + + public ValueTask HandleAsync( + ExecutorBehaviorContext context, + ExecutorBehaviorContinuation continuation, + CancellationToken cancellationToken) => + new ValueTask(this._result); + } + + private sealed class FaultyBehavior : IExecutorBehavior + { + public ValueTask HandleAsync( + ExecutorBehaviorContext context, + ExecutorBehaviorContinuation continuation, + CancellationToken cancellationToken) => + throw new InvalidOperationException("Intentional behavior failure for testing"); + } + + private sealed class CapturingWorkflowBehavior : IWorkflowBehavior + { + private readonly Action _capture; + + public CapturingWorkflowBehavior(Action capture) + { + this._capture = capture; + } + + public async ValueTask HandleAsync( + WorkflowBehaviorContext context, + WorkflowBehaviorContinuation continuation, + CancellationToken cancellationToken) + { + if (context.Stage == WorkflowStage.Starting) + { + this._capture(context); + } + + return await continuation(cancellationToken); + } + } + + private sealed class CapturingExecutorBehavior : IExecutorBehavior + { + private readonly Action _capture; + + public CapturingExecutorBehavior(Action capture) + { + this._capture = capture; + } + + public async ValueTask HandleAsync( + ExecutorBehaviorContext context, + ExecutorBehaviorContinuation continuation, + CancellationToken cancellationToken) + { + this._capture(context); + return await continuation(cancellationToken); + } + } +} diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Behaviors/WorkflowBehaviorOptionsTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Behaviors/WorkflowBehaviorOptionsTests.cs new file mode 100644 index 00000000000..2460d3de96e --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Behaviors/WorkflowBehaviorOptionsTests.cs @@ -0,0 +1,296 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Threading; +using System.Threading.Tasks; +using FluentAssertions; +using Microsoft.Agents.AI.Workflows.Behaviors; + +namespace Microsoft.Agents.AI.Workflows.UnitTests.Behaviors; + +/// +/// Tests for the WorkflowBehaviorOptions API and registration mechanisms. +/// +public class WorkflowBehaviorOptionsTests +{ + [Fact] + public void AddExecutorBehavior_WithInstance_RegistersBehavior() + { + // Arrange + var options = new WorkflowBehaviorOptions(); + var behavior = new TestExecutorBehavior(); + + // Act + options.AddExecutorBehavior(behavior); + var pipeline = options.BuildPipeline(); + + // Assert + pipeline.Should().NotBeNull(); + pipeline!.HasExecutorBehaviors.Should().BeTrue(); + } + + [Fact] + public void AddWorkflowBehavior_WithInstance_RegistersBehavior() + { + // Arrange + var options = new WorkflowBehaviorOptions(); + var behavior = new TestWorkflowBehavior(); + + // Act + options.AddWorkflowBehavior(behavior); + var pipeline = options.BuildPipeline(); + + // Assert + pipeline.Should().NotBeNull(); + pipeline!.HasWorkflowBehaviors.Should().BeTrue(); + } + + [Fact] + public void AddExecutorBehavior_MultipleInstances_RegistersAllBehaviors() + { + // Arrange + var options = new WorkflowBehaviorOptions(); + var behavior1 = new TestExecutorBehavior(); + var behavior2 = new TestExecutorBehavior(); + var behavior3 = new TestExecutorBehavior(); + + // Act + options.AddExecutorBehavior(behavior1); + options.AddExecutorBehavior(behavior2); + options.AddExecutorBehavior(behavior3); + var pipeline = options.BuildPipeline(); + + // Assert + pipeline.Should().NotBeNull(); + pipeline!.HasExecutorBehaviors.Should().BeTrue(); + } + + [Fact] + public void AddWorkflowBehavior_MultipleInstances_RegistersAllBehaviors() + { + // Arrange + var options = new WorkflowBehaviorOptions(); + var behavior1 = new TestWorkflowBehavior(); + var behavior2 = new TestWorkflowBehavior(); + + // Act + options.AddWorkflowBehavior(behavior1); + options.AddWorkflowBehavior(behavior2); + var pipeline = options.BuildPipeline(); + + // Assert + pipeline.Should().NotBeNull(); + pipeline!.HasWorkflowBehaviors.Should().BeTrue(); + } + + [Fact] + public void BuildPipeline_WithNoBehaviors_ReturnsEmptyPipeline() + { + // Arrange + var options = new WorkflowBehaviorOptions(); + + // Act + var pipeline = options.BuildPipeline(); + + // Assert + pipeline.Should().NotBeNull(); + pipeline!.HasExecutorBehaviors.Should().BeFalse(); + pipeline.HasWorkflowBehaviors.Should().BeFalse(); + } + + [Fact] + public void WorkflowBuilder_WithBehaviors_ConfiguresBehaviors() + { + // Arrange + var behavior = new TestExecutorBehavior(); + var executor = new SimpleExecutor("test"); + + // Act + var workflow = new WorkflowBuilder(executor) + .WithBehaviors(options => options.AddExecutorBehavior(behavior)) + .Build(); + + // Assert + workflow.Should().NotBeNull(); + workflow.BehaviorPipeline.Should().NotBeNull(); + workflow.BehaviorPipeline!.HasExecutorBehaviors.Should().BeTrue(); + } + + [Fact] + public void WorkflowBuilder_WithBehaviors_SupportsFluentAPI() + { + // Arrange + var executor = new SimpleExecutor("test"); + + // Act + var workflow = new WorkflowBuilder(executor) + .WithBehaviors(options => + { + options.AddExecutorBehavior(new TestExecutorBehavior()); + options.AddWorkflowBehavior(new TestWorkflowBehavior()); + }) + .Build(); + + // Assert + workflow.Should().NotBeNull(); + workflow.BehaviorPipeline.Should().NotBeNull(); + workflow.BehaviorPipeline!.HasExecutorBehaviors.Should().BeTrue(); + workflow.BehaviorPipeline.HasWorkflowBehaviors.Should().BeTrue(); + } + + [Fact] + public void WorkflowBuilder_WithoutBehaviors_HasNullPipeline() + { + // Arrange + var executor = new SimpleExecutor("test"); + + // Act + var workflow = new WorkflowBuilder(executor).Build(); + + // Assert + workflow.Should().NotBeNull(); + workflow.BehaviorPipeline.Should().BeNull(); + } + + [Fact] + public void AddExecutorBehavior_NullBehavior_ThrowsArgumentNullException() + { + // Arrange + var options = new WorkflowBehaviorOptions(); + + // Act + Action act = () => options.AddExecutorBehavior(null!); + + // Assert + act.Should().Throw(); + } + + [Fact] + public void AddWorkflowBehavior_NullBehavior_ThrowsArgumentNullException() + { + // Arrange + var options = new WorkflowBehaviorOptions(); + + // Act + Action act = () => options.AddWorkflowBehavior(null!); + + // Assert + act.Should().Throw(); + } + + [Fact] + public void AddExecutorBehavior_GenericOverload_RegistersBehavior() + { + // Arrange + var options = new WorkflowBehaviorOptions(); + + // Act + options.AddExecutorBehavior(); + var pipeline = options.BuildPipeline(); + + // Assert + pipeline.Should().NotBeNull(); + pipeline!.HasExecutorBehaviors.Should().BeTrue(); + } + + [Fact] + public void AddWorkflowBehavior_GenericOverload_RegistersBehavior() + { + // Arrange + var options = new WorkflowBehaviorOptions(); + + // Act + options.AddWorkflowBehavior(); + var pipeline = options.BuildPipeline(); + + // Assert + pipeline.Should().NotBeNull(); + pipeline!.HasWorkflowBehaviors.Should().BeTrue(); + } + + [Fact] + public void AddBehavior_ReturnsSameOptionsInstance_ForChaining() + { + // Arrange + var options = new WorkflowBehaviorOptions(); + + // Act - every registration overload is documented as chainable + var chained = options + .AddExecutorBehavior(new TestExecutorBehavior()) + .AddWorkflowBehavior(new TestWorkflowBehavior()) + .AddExecutorBehavior() + .AddWorkflowBehavior(); + + // Assert + chained.Should().BeSameAs(options); + var pipeline = options.BuildPipeline(); + pipeline!.HasExecutorBehaviors.Should().BeTrue(); + pipeline.HasWorkflowBehaviors.Should().BeTrue(); + } + + [Fact] + public void WorkflowBuilder_WithBehaviors_NullConfigure_ThrowsArgumentNullException() + { + // Arrange + var builder = new WorkflowBuilder(new SimpleExecutor("test")); + + // Act + Action act = () => builder.WithBehaviors(null!); + + // Assert - a null configure callback must not be silently ignored + act.Should().Throw(); + } + + [Fact] + public void WorkflowBuilder_WithBehaviors_CalledTwice_AccumulatesBehaviors() + { + // Arrange + var executor = new SimpleExecutor("test"); + + // Act - successive calls share one options instance rather than replacing it + var workflow = new WorkflowBuilder(executor) + .WithBehaviors(options => options.AddExecutorBehavior(new TestExecutorBehavior())) + .WithBehaviors(options => options.AddWorkflowBehavior(new TestWorkflowBehavior())) + .Build(); + + // Assert + workflow.BehaviorPipeline.Should().NotBeNull(); + workflow.BehaviorPipeline!.HasExecutorBehaviors.Should().BeTrue(); + workflow.BehaviorPipeline.HasWorkflowBehaviors.Should().BeTrue(); + } + + // Test helper classes + private sealed class TestExecutorBehavior : IExecutorBehavior + { + public async ValueTask HandleAsync( + ExecutorBehaviorContext context, + ExecutorBehaviorContinuation continuation, + CancellationToken cancellationToken) + { + return await continuation(cancellationToken); + } + } + + private sealed class TestWorkflowBehavior : IWorkflowBehavior + { + public async ValueTask HandleAsync( + WorkflowBehaviorContext context, + WorkflowBehaviorContinuation continuation, + CancellationToken cancellationToken) + { + return await continuation(cancellationToken); + } + } + + private sealed class SimpleExecutor : Executor + { + public SimpleExecutor(string id) : base(id) { } + + protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder) => + protocolBuilder.ConfigureRoutes(routeBuilder => routeBuilder.AddHandler(async (message, context) => + { + await context.SendMessageAsync(message); + return message; + })); + } +}