diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/StreamingRunEventStream.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/StreamingRunEventStream.cs index e6afbb5440..ca0cc52641 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/StreamingRunEventStream.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/StreamingRunEventStream.cs @@ -2,10 +2,12 @@ using System; using System.Collections.Generic; +using System.Diagnostics; using System.Runtime.CompilerServices; using System.Threading; using System.Threading.Channels; using System.Threading.Tasks; +using Microsoft.Agents.AI.Workflows.Observability; namespace Microsoft.Agents.AI.Workflows.Execution; @@ -15,6 +17,9 @@ namespace Microsoft.Agents.AI.Workflows.Execution; /// internal sealed class StreamingRunEventStream : IRunEventStream { + private static readonly string s_namespace = typeof(StreamingRunEventStream).Namespace!; + private static readonly ActivitySource s_activitySource = new(s_namespace); + private readonly Channel _eventChannel; private readonly ISuperStepRunner _stepRunner; private readonly InputWaiter _inputWaiter; @@ -58,6 +63,9 @@ internal sealed class StreamingRunEventStream : IRunEventStream // Subscribe to events - they will flow directly to the channel as they're raised this._stepRunner.OutgoingEvents.EventRaised += OnEventRaisedAsync; + using Activity? activity = s_activitySource.StartActivity(ActivityNames.WorkflowRun); + activity?.SetTag(Tags.WorkflowId, this._stepRunner.StartExecutorId).SetTag(Tags.RunId, this._stepRunner.RunId); + try { // Wait for the first input before starting @@ -65,6 +73,7 @@ internal sealed class StreamingRunEventStream : IRunEventStream await this._inputWaiter.WaitForInputAsync(cancellationToken: linkedSource.Token).ConfigureAwait(false); this._runStatus = RunStatus.Running; + activity?.AddEvent(new ActivityEvent(EventNames.WorkflowStarted)); while (!linkedSource.Token.IsCancellationRequested) { @@ -99,9 +108,17 @@ internal sealed class StreamingRunEventStream : IRunEventStream { // Expected during shutdown } - catch (Exception e) + catch (Exception ex) { - await this._eventChannel.Writer.WriteAsync(new WorkflowErrorEvent(e), linkedSource.Token).ConfigureAwait(false); + if (activity != null) + { + activity.AddEvent(new ActivityEvent(EventNames.WorkflowError, tags: new() { + { Tags.ErrorType, ex.GetType().FullName }, + { Tags.BuildErrorMessage, ex.Message }, + })); + activity.CaptureException(ex); + } + await this._eventChannel.Writer.WriteAsync(new WorkflowErrorEvent(ex), linkedSource.Token).ConfigureAwait(false); } finally { @@ -110,6 +127,7 @@ internal sealed class StreamingRunEventStream : IRunEventStream // Mark as ended when run loop exits this._runStatus = RunStatus.Ended; + activity?.AddEvent(new ActivityEvent(EventNames.WorkflowCompleted)); } async ValueTask OnEventRaisedAsync(object? sender, WorkflowEvent e) diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/ObservabilityTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/ObservabilityTests.cs new file mode 100644 index 0000000000..7101ad13d4 --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/ObservabilityTests.cs @@ -0,0 +1,186 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Diagnostics; +using System.Linq; +using System.Threading.Tasks; +using FluentAssertions; +using Microsoft.Agents.AI.Workflows.InProc; +using Microsoft.Agents.AI.Workflows.Observability; + +namespace Microsoft.Agents.AI.Workflows.UnitTests; + +/// +/// These tests ensure that OpenTelemetry Activity traces are properly created for workflow monitoring. +/// Tests are run in a collection to avoid parallel execution since ActivityListener is global. +/// Each test creates a new instance of ObservabilityTests and runs in serial within the collection. +/// This prevents interference between tests due to the global nature of ActivityListener. +/// +[Collection("ObservabilityTests")] +public sealed class ObservabilityTests : IDisposable +{ + private readonly ActivityListener _activityListener; + private readonly ConcurrentBag _capturedActivities = []; + + private bool _isDisposed; + + public ObservabilityTests() + { + // Set up activity listener to capture activities from workflow + // This is global and captures ALL workflow activities from ANY test in the same process! + this._activityListener = new ActivityListener + { + ShouldListenTo = source => source.Name.Contains(typeof(Workflow).Namespace!), + Sample = (ref ActivityCreationOptions options) => ActivitySamplingResult.AllData, + ActivityStarted = activity => this._capturedActivities.Add(activity), + }; + ActivitySource.AddActivityListener(this._activityListener); + } + + /// + /// Create a sample workflow for testing. + /// + /// + /// This workflow is expected to create 8 activities that will be captured by the tests + /// - ActivityNames.WorkflowBuild + /// - ActivityNames.WorkflowRun + /// -- ActivityNames.EdgeGroupProcess + /// -- ActivityNames.ExecutorProcess (UppercaseExecutor) + /// --- ActivityNames.MessageSend + /// ---- ActivityNames.EdgeGroupProcess + /// -- ActivityNames.ExecutorProcess (ReverseTextExecutor) + /// --- ActivityNames.MessageSend + /// + /// The created workflow. + private static Workflow CreateWorkflow() + { + // Create the executors + Func uppercaseFunc = s => s.ToUpperInvariant(); + var uppercase = uppercaseFunc.BindAsExecutor("UppercaseExecutor"); + + Func reverseFunc = s => new string(s.Reverse().ToArray()); + var reverse = reverseFunc.BindAsExecutor("ReverseTextExecutor"); + + // Build the workflow by connecting executors sequentially + WorkflowBuilder builder = new(uppercase); + builder.AddEdge(uppercase, reverse).WithOutputFrom(reverse); + + return builder.Build(); + } + + private static Dictionary GetExpectedActivityNameCounts() => + new() + { + { ActivityNames.WorkflowBuild, 1 }, + { ActivityNames.WorkflowRun, 1 }, + { ActivityNames.EdgeGroupProcess, 2 }, + { ActivityNames.ExecutorProcess, 2 }, + { ActivityNames.MessageSend, 2 } + }; + + private static InProcessExecutionEnvironment GetExecutionEnvironment(string name) => + name switch + { + "Default" => InProcessExecution.Default, + "Lockstep" => InProcessExecution.Lockstep, + "OffThread" => InProcessExecution.OffThread, + "Concurrent" => InProcessExecution.Concurrent, + _ => throw new ArgumentException($"Unknown execution environment name: {name}") + }; + + public void Dispose() + { + if (!this._isDisposed) + { + this._activityListener?.Dispose(); + this._isDisposed = true; + } + } + + private async Task TestWorkflowEndToEndActivitiesAsync(string executionEnvironmentName) + { + // Arrange + // Create a test activity to correlate captured activities + using var testActivity = new Activity("ObservabilityTest").Start(); + + // Act + var workflow = CreateWorkflow(); + var executionEnvironment = GetExecutionEnvironment(executionEnvironmentName); + Run run = await executionEnvironment.RunAsync(workflow, "Hello, World!"); + await run.DisposeAsync(); + + await Task.Delay(100); // Allow time for activities to be captured + + // Assert + var capturedActivities = this._capturedActivities.Where(a => a.RootId == testActivity.RootId).ToList(); + capturedActivities.Should().HaveCount(8, "Exactly 8 activities should be created."); + + // Make sure all expected activities exist and have the correct count + foreach (var kvp in GetExpectedActivityNameCounts()) + { + var activityName = kvp.Key; + var expectedCount = kvp.Value; + var actualCount = capturedActivities.Count(a => a.OperationName == activityName); + actualCount.Should().Be(expectedCount, $"Activity '{activityName}' should occur {expectedCount} times."); + } + + // Verify WorkflowRun activity events include workflow lifecycle events + var workflowRunActivity = capturedActivities.First(a => a.OperationName == ActivityNames.WorkflowRun); + var activityEvents = workflowRunActivity.Events.ToList(); + activityEvents.Should().Contain(e => e.Name == EventNames.WorkflowStarted, "activity should have workflow started event"); + activityEvents.Should().Contain(e => e.Name == EventNames.WorkflowCompleted, "activity should have workflow completed event"); + } + + [Fact] + public async Task CreatesWorkflowEndToEndActivities_WithCorrectName_DefaultAsync() + { + await this.TestWorkflowEndToEndActivitiesAsync("Default"); + } + + [Fact] + public async Task CreatesWorkflowEndToEndActivities_WithCorrectName_OffThreadAsync() + { + await this.TestWorkflowEndToEndActivitiesAsync("OffThread"); + } + + [Fact] + public async Task CreatesWorkflowEndToEndActivities_WithCorrectName_ConcurrentAsync() + { + await this.TestWorkflowEndToEndActivitiesAsync("Concurrent"); + } + + [Fact] + public async Task CreatesWorkflowEndToEndActivities_WithCorrectName_LockstepAsync() + { + await this.TestWorkflowEndToEndActivitiesAsync("Lockstep"); + } + + [Fact] + public async Task CreatesWorkflowActivities_WithCorrectNameAsync() + { + // Arrange + // Create a test activity to correlate captured activities + using var testActivity = new Activity("ObservabilityTest").Start(); + + // Act + CreateWorkflow(); + await Task.Delay(100); // Allow time for activities to be captured + + // Assert + var capturedActivities = this._capturedActivities.Where(a => a.RootId == testActivity.RootId).ToList(); + capturedActivities.Should().HaveCount(1, "Exactly 1 activity should be created."); + capturedActivities[0].OperationName.Should().Be(ActivityNames.WorkflowBuild, + "The activity should have the correct operation name for workflow build."); + + var events = capturedActivities[0].Events.ToList(); + events.Should().Contain(e => e.Name == EventNames.BuildStarted, "activity should have build started event"); + events.Should().Contain(e => e.Name == EventNames.BuildValidationCompleted, "activity should have build validation completed event"); + events.Should().Contain(e => e.Name == EventNames.BuildCompleted, "activity should have build completed event"); + + var tags = capturedActivities[0].Tags.ToDictionary(t => t.Key, t => t.Value); + tags.Should().ContainKey(Tags.WorkflowId); + tags.Should().ContainKey(Tags.WorkflowDefinition); + } +}