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

.NET: Make sure Workflow activities are as expected (#1903)

Browse Source 
* Make sure Workflow activities are as expected

* misc

* Copliot comments

* Fix unit tests

* Improve test stability

* Fix unit tests

* Fix formatting
This commit is contained in:
Tao Chen
2025-11-04 16:37:31 -08:00
committed by GitHub
Unverified
parent 2499262f30
commit 552f7c781d
2 changed files with 206 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dotnet/src/Microsoft.Agents.AI.Workflows/Execution/StreamingRunEventStream.cs
+20 -2
View File
@@ -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;
/// </summary>
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<WorkflowEvent> _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)