.NET: Fixing issue where OpenTelemetry span is never exported in .NET in-process workflow execution

[alliscode]

This pull request addresses the issue where workflow run telemetry spans (Activity objects) were not always properly stopped and exported, particularly in streaming and lockstep execution environments. The changes ensure that workflow run activities are disposed as soon as the workflow reaches the idle state or when the run loop exits, preventing telemetry data from being lost. Additionally, comprehensive regression tests are added to verify correct activity lifecycle management.

Improvements to Activity Lifecycle Management:

• Ensured that the workflow.run Activity is disposed immediately when the workflow reaches the Idle state, so telemetry spans are promptly exported rather than waiting for cancellation or disposal.
• Added a safety net to dispose of the workflow.run Activity if it was not already stopped when the run loop exits, covering cancellation and error scenarios.
• Removed the using statement from the activity initialization to allow manual control over the activity's disposal timing.

Testing and Regression Coverage:

• Added a new test file WorkflowRunActivityStopTests.cs to verify that workflow run activities are always properly stopped and exported to telemetry backends, covering lockstep, off-thread, and streaming execution environments, as well as ensuring that all started activities are stopped.

Closes #4155

Contribution Checklist

• The code builds clean without any errors or warnings
• The PR follows the Contribution Guidelines
• All unit tests pass, and I have added new tests where possible
• Is this a breaking change? If yes, add "[BREAKING]" prefix to the title of the PR.

[Copilot]

Pull request overview

This PR aims to ensure OpenTelemetry workflow-run spans (Activity) are reliably stopped/disposed (and therefore exported) during .NET in-process workflow execution, including streaming scenarios, and adds regression tests around activity lifecycle behavior.

Changes:

• Updated StreamingRunEventStream.RunLoopAsync to manually manage the workflow-run Activity lifecycle (stop on Idle and ensure disposal on loop exit).
• Added WorkflowRunActivityStopTests to assert workflow-run activities are started and stopped across multiple execution modes.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 3 comments.

File	Description
dotnet/src/Microsoft.Agents.AI.Workflows/Execution/StreamingRunEventStream.cs	Changes workflow-run Activity disposal timing to stop/export spans earlier and adds a safety-net disposal on exit.
dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/WorkflowRunActivityStopTests.cs	Adds regression coverage validating workflow-run activities are stopped/disposed in lockstep, off-thread, and streaming usage.

[Copilot]

Pull request overview

Copilot reviewed 7 out of 7 changed files in this pull request and generated 7 comments.

[alliscode]

@copilot open a new pull request to apply changes based on the comments in this thread

[alliscode]

@copilot open a new pull request to apply changes based on the comments in this thread

[Copilot]

Pull request overview

Copilot reviewed 8 out of 8 changed files in this pull request and generated 2 comments.

Comments suppressed due to low confidence (1)

dotnet/src/Microsoft.Agents.AI.Workflows/Execution/StreamingRunEventStream.cs:123

• In RunLoopAsync, a new runActivity is created on every iteration of the outer while loop, but InputWaiter.WaitForInputAsync(TimeSpan.FromSeconds(1), ...) returns after the timeout even when no input was signaled. This can emit repeated workflow_invoke spans (and flip _runStatus back to Running) during idle periods with no new input, creating noisy/incorrect telemetry and potential performance issues. Consider having WaitForInputAsync surface whether it was actually signaled (e.g., return the bool from SemaphoreSlim.WaitAsync(timeout, token)), and only start a new run-stage activity / set _runStatus = Running when real input arrives (or when there are unprocessed messages).

            while (!linkedSource.Token.IsCancellationRequested)
            {
                // Start a new run-stage activity for this input→processing→halt cycle
                runActivity = this._stepRunner.TelemetryContext.StartWorkflowRunActivity();
                runActivity?.SetTag(Tags.WorkflowId, this._stepRunner.StartExecutorId)
                            .SetTag(Tags.SessionId, this._stepRunner.SessionId);
                runActivity?.AddEvent(new ActivityEvent(EventNames.WorkflowStarted));

                // Run all available supersteps continuously
                // Events are streamed out in real-time as they happen via the event handler
                while (this._stepRunner.HasUnprocessedMessages && !linkedSource.Token.IsCancellationRequested)
                {
                    await this._stepRunner.RunSuperStepAsync(linkedSource.Token).ConfigureAwait(false);
                }

                // Update status based on what's waiting
                this._runStatus = this._stepRunner.HasUnservicedRequests
                    ? RunStatus.PendingRequests
                    : RunStatus.Idle;

                // Signal completion to consumer so they can check status and decide whether to continue
                // Increment epoch so next consumer iteration gets a new completion signal
                // Capture the status at this moment to avoid race conditions with event reading
                int currentEpoch = Interlocked.Increment(ref this._completionEpoch);
                RunStatus capturedStatus = this._runStatus;
                await this._eventChannel.Writer.WriteAsync(new InternalHaltSignal(currentEpoch, capturedStatus), linkedSource.Token).ConfigureAwait(false);

                // Close the run-stage activity when processing halts.
                // A new run activity will be created when the next input arrives.
                if (runActivity is not null)
                {
                    runActivity.AddEvent(new ActivityEvent(EventNames.WorkflowCompleted));
                    runActivity.Dispose();
                    runActivity = null;
                }

                // Wait for next input from the consumer
                // Works for both Idle (no work) and PendingRequests (waiting for responses)
                await this._inputWaiter.WaitForInputAsync(TimeSpan.FromSeconds(1), linkedSource.Token).ConfigureAwait(false);

                // When signaled, resume running
                this._runStatus = RunStatus.Running;
            }

[lokitoth]

I wonder if there is stitching that we need to do across Checkpoint restore boundaries...