Skip to content

Commit ae784bf

Browse files
author
alliscode
committed
Address Copilot review: fix Activity/CTS disposal, rename activity, add error tag\n\n1. LockstepRunEventStream: Remove 'using' from Activity in async iterator\n and manually dispose in finally block (fixes microsoft#4155 pattern). Also dispose\n linkedSource CTS in finally to prevent leak.\n2. Tags.cs: Add ErrorMessage (\"error.message\") tag for runtime errors,\n distinct from BuildErrorMessage (\"build.error.message\").\n3. ActivityNames: Rename WorkflowRun from \"workflow_invoke\" to \"workflow.run\"\n for cross-language consistency.\n4. WorkflowTelemetryContext: Fix XML doc to say \"outer/parent span\" instead\n of \"root-level span\".\n5. ObservabilityTests: Assert WorkflowSession absence when DisableWorkflowRun\n is true.\n6. WorkflowRunActivityStopTests: Fix streaming test race by disposing\n StreamingRun before asserting activities are stopped.\n7. StreamingRunEventStream/LockstepRunEventStream: Use Tags.ErrorMessage\n instead of Tags.BuildErrorMessage for runtime error events."
1 parent bf80930 commit ae784bf

7 files changed

Lines changed: 28 additions & 11 deletions

File tree

‎dotnet/src/Microsoft.Agents.AI.Workflows/Execution/LockstepRunEventStream.cs‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -56,7 +56,11 @@ public async IAsyncEnumerable<WorkflowEvent> TakeEventStreamAsync(bool blockOnPe
5656

5757
this._stepRunner.OutgoingEvents.EventRaised += OnWorkflowEventAsync;
5858

59-
using Activity? activity = this._stepRunner.TelemetryContext.StartWorkflowRunActivity();
59+
// Do NOT use a 'using' declaration for the Activity inside an async iterator.
60+
// In compiled async iterator state machines, 'using' locals are only disposed when
61+
// DisposeAsync() is called on the enumerator. Activity.Stop (fired by Dispose) is
62+
// time-sensitive for OpenTelemetry export, so we dispose explicitly in the finally block.
63+
Activity? activity = this._stepRunner.TelemetryContext.StartWorkflowRunActivity();
6064
activity?.SetTag(Tags.WorkflowId, this._stepRunner.StartExecutorId).SetTag(Tags.SessionId, this._stepRunner.SessionId);
6165

6266
try
@@ -85,7 +89,7 @@ public async IAsyncEnumerable<WorkflowEvent> TakeEventStreamAsync(bool blockOnPe
8589
{
8690
activity.AddEvent(new ActivityEvent(EventNames.WorkflowError, tags: new() {
8791
{ Tags.ErrorType, ex.GetType().FullName },
88-
{ Tags.BuildErrorMessage, ex.Message },
92+
{ Tags.ErrorMessage, ex.Message },
8993
}));
9094
activity.CaptureException(ex);
9195
throw;
@@ -141,6 +145,11 @@ public async IAsyncEnumerable<WorkflowEvent> TakeEventStreamAsync(bool blockOnPe
141145
{
142146
this.RunStatus = this._stepRunner.HasUnservicedRequests ? RunStatus.PendingRequests : RunStatus.Idle;
143147
this._stepRunner.OutgoingEvents.EventRaised -= OnWorkflowEventAsync;
148+
149+
// Explicitly dispose the Activity so Activity.Stop fires deterministically,
150+
// regardless of how the async iterator enumerator is disposed.
151+
activity?.Dispose();
152+
linkedSource.Dispose();
144153
}
145154

146155
ValueTask OnWorkflowEventAsync(object? sender, WorkflowEvent e)

‎dotnet/src/Microsoft.Agents.AI.Workflows/Execution/StreamingRunEventStream.cs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -132,7 +132,7 @@ private async Task RunLoopAsync(CancellationToken cancellationToken)
132132
{
133133
runActivity.AddEvent(new ActivityEvent(EventNames.WorkflowError, tags: new() {
134134
{ Tags.ErrorType, ex.GetType().FullName },
135-
{ Tags.BuildErrorMessage, ex.Message },
135+
{ Tags.ErrorMessage, ex.Message },
136136
}));
137137
runActivity.CaptureException(ex);
138138
}
@@ -142,7 +142,7 @@ private async Task RunLoopAsync(CancellationToken cancellationToken)
142142
{
143143
sessionActivity.AddEvent(new ActivityEvent(EventNames.SessionError, tags: new() {
144144
{ Tags.ErrorType, ex.GetType().FullName },
145-
{ Tags.BuildErrorMessage, ex.Message },
145+
{ Tags.ErrorMessage, ex.Message },
146146
}));
147147
sessionActivity.CaptureException(ex);
148148
}
@@ -157,7 +157,7 @@ private async Task RunLoopAsync(CancellationToken cancellationToken)
157157
// Mark as ended when run loop exits
158158
this._runStatus = RunStatus.Ended;
159159

160-
// Safety net: stop the run-stage activity if not already stopped (e.g. on cancellation or error)
160+
// Stop the run-stage activity if not already stopped (e.g. on cancellation or error)
161161
if (runActivity is not null)
162162
{
163163
runActivity.AddEvent(new ActivityEvent(EventNames.WorkflowCompleted));

‎dotnet/src/Microsoft.Agents.AI.Workflows/Observability/ActivityNames.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@ internal static class ActivityNames
66
{
77
public const string WorkflowBuild = "workflow.build";
88
public const string WorkflowSession = "workflow.session";
9-
public const string WorkflowRun = "workflow_invoke";
9+
public const string WorkflowRun = "workflow.run";
1010
public const string MessageSend = "message.send";
1111
public const string ExecutorProcess = "executor.process";
1212
public const string EdgeGroupProcess = "edge_group.process";

‎dotnet/src/Microsoft.Agents.AI.Workflows/Observability/Tags.cs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ internal static class Tags
1111
public const string BuildErrorMessage = "build.error.message";
1212
public const string BuildErrorType = "build.error.type";
1313
public const string ErrorType = "error.type";
14+
public const string ErrorMessage = "error.message";
1415
public const string SessionId = "session.id";
1516
public const string ExecutorId = "executor.id";
1617
public const string ExecutorType = "executor.type";

‎dotnet/src/Microsoft.Agents.AI.Workflows/Observability/WorkflowTelemetryContext.cs‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -88,9 +88,10 @@ public WorkflowTelemetryContext(WorkflowTelemetryOptions options, ActivitySource
8888
}
8989

9090
/// <summary>
91-
/// Starts a workflow session activity if enabled. This is a root-level span
91+
/// Starts a workflow session activity if enabled. This is the outer/parent span
9292
/// that represents the entire lifetime of a workflow execution (from start
93-
/// until stop, cancellation, or error). Individual run stages are nested within it.
93+
/// until stop, cancellation, or error) within the current trace.
94+
/// Individual run stages are typically nested within it.
9495
/// </summary>
9596
/// <returns>An activity if workflow run telemetry is enabled, otherwise null.</returns>
9697
public Activity? StartWorkflowSessionActivity()

‎dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/ObservabilityTests.cs‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -277,6 +277,9 @@ public async Task DisableWorkflowRun_PreventsWorkflowRunActivityAsync()
277277
capturedActivities.Should().NotContain(
278278
a => a.OperationName.StartsWith(ActivityNames.WorkflowRun, StringComparison.Ordinal),
279279
"WorkflowRun activity should be disabled.");
280+
capturedActivities.Should().NotContain(
281+
a => a.OperationName.StartsWith(ActivityNames.WorkflowSession, StringComparison.Ordinal),
282+
"WorkflowSession activity should also be disabled when DisableWorkflowRun is true.");
280283
capturedActivities.Should().Contain(
281284
a => a.OperationName.StartsWith(ActivityNames.WorkflowBuild, StringComparison.Ordinal),
282285
"Other activities should still be created.");

‎dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/WorkflowRunActivityStopTests.cs‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -2,12 +2,10 @@
22

33
using System;
44
using System.Collections.Concurrent;
5-
using System.Collections.Generic;
65
using System.Diagnostics;
76
using System.Linq;
87
using System.Threading.Tasks;
98
using FluentAssertions;
10-
using Microsoft.Agents.AI.Workflows.InProc;
119
using Microsoft.Agents.AI.Workflows.Observability;
1210

1311
namespace Microsoft.Agents.AI.Workflows.UnitTests;
@@ -166,12 +164,17 @@ public async Task WorkflowRunActivity_IsStopped_Streaming_OffThreadAsync()
166164

167165
// Act - use streaming path (WatchStreamAsync), which is the pattern from the issue
168166
var workflow = CreateWorkflow();
169-
await using StreamingRun run = await InProcessExecution.OffThread.RunStreamingAsync(workflow, "Hello, World!");
167+
StreamingRun run = await InProcessExecution.OffThread.RunStreamingAsync(workflow, "Hello, World!");
170168
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
171169
{
172170
// Consume all events
173171
}
174172

173+
// Dispose the run before asserting — the run Activity is disposed when the
174+
// run loop exits, which happens during DisposeAsync. Without this, assertions
175+
// can race against the background run loop's finally block.
176+
await run.DisposeAsync();
177+
175178
// Assert - workflow.session should have been started
176179
var startedSessions = this._startedActivities
177180
.Where(a => a.RootId == testActivity.RootId &&

0 commit comments

Comments
 (0)