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

Fix streaming run double disposal in tests and lockstep republishing before Started event is emitted.

Browse Source
This commit is contained in:
Peter Ibekwe
2026-03-27 15:06:23 -07:00
Unverified
parent 337e8e6897
commit 8961c474f5
2 changed files with 66 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dotnet/src/Microsoft.Agents.AI.Workflows/Execution/LockstepRunEventStream.cs
+5 -4
View File
@@ -59,10 +59,6 @@ internal sealed class LockstepRunEventStream : IRunEventStream
using CancellationTokenSource linkedSource = CancellationTokenSource.CreateLinkedTokenSource(this._stopCancellation.Token, cancellationToken);
// Re-emit any pending external requests that were restored from a checkpoint
// before this subscription was active. For non-resume starts this is a no-op.
await this._stepRunner.RepublishPendingEventsAsync(linkedSource.Token).ConfigureAwait(false);
// Re-establish session as parent so the run activity nests correctly.
Activity.Current = this._sessionActivity;
@@ -78,6 +74,11 @@ internal sealed class LockstepRunEventStream : IRunEventStream
// Emit WorkflowStartedEvent to the event stream for consumers
this._eventSink.Enqueue(new WorkflowStartedEvent());
// Re-emit any pending external requests that were restored from a checkpoint
// before this subscription was active. For non-resume starts this is a no-op.
// This runs after WorkflowStartedEvent so consumers always see the started event first.
await this._stepRunner.RepublishPendingEventsAsync(linkedSource.Token).ConfigureAwait(false);
// When resuming from a checkpoint with only pending requests (no queued messages),
// the inner processing loop won't execute, so we must drain events now.
// For normal starts this is a no-op since the inner loop handles the drain.
dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/CheckpointResumeTests.cs
+61 -58
View File
@@ -38,31 +38,29 @@ public class CheckpointResumeTests
InProcessExecutionEnvironment env = environment.ToWorkflowExecutionEnvironment();
// Act 1: Run workflow, collect pending requests and a checkpoint.
await using StreamingRun firstRun = await env.WithCheckpointing(checkpointManager)
.RunStreamingAsync(workflow, "Hello");
List<ExternalRequest> originalRequests = [];
CheckpointInfo? checkpoint = null;
await foreach (WorkflowEvent evt in firstRun.WatchStreamAsync(blockOnPendingRequest: false))
await using (StreamingRun firstRun = await env.WithCheckpointing(checkpointManager)
.RunStreamingAsync(workflow, "Hello"))
{
if (evt is RequestInfoEvent requestInfo)
await foreach (WorkflowEvent evt in firstRun.WatchStreamAsync(blockOnPendingRequest: false))
{
originalRequests.Add(requestInfo.Request);
if (evt is RequestInfoEvent requestInfo)
{
originalRequests.Add(requestInfo.Request);
}
if (evt is SuperStepCompletedEvent step && step.CompletionInfo?.Checkpoint is { } cp)
{
checkpoint = cp;
}
}
if (evt is SuperStepCompletedEvent step && step.CompletionInfo?.Checkpoint is { } cp)
{
checkpoint = cp;
}
originalRequests.Should().NotBeEmpty("the workflow should have created at least one external request");
checkpoint.Should().NotBeNull("a checkpoint should have been created");
}
originalRequests.Should().NotBeEmpty("the workflow should have created at least one external request");
checkpoint.Should().NotBeNull("a checkpoint should have been created");
// Dispose the first run to release ownership before resuming.
await firstRun.DisposeAsync();
// Act 2: Resume from the checkpoint.
await using StreamingRun resumed = await env.WithCheckpointing(checkpointManager)
.ResumeStreamingAsync(workflow, checkpoint!);
@@ -107,20 +105,21 @@ public class CheckpointResumeTests
InProcessExecutionEnvironment env = environment.ToWorkflowExecutionEnvironment();
// First run: collect a checkpoint with pending requests.
await using StreamingRun firstRun = await env.WithCheckpointing(checkpointManager)
.RunStreamingAsync(workflow, "Hello");
CheckpointInfo? checkpoint = null;
await foreach (WorkflowEvent evt in firstRun.WatchStreamAsync(blockOnPendingRequest: false))
{
if (evt is SuperStepCompletedEvent step && step.CompletionInfo?.Checkpoint is { } cp)
{
checkpoint = cp;
}
}
checkpoint.Should().NotBeNull();
await firstRun.DisposeAsync();
await using (StreamingRun firstRun = await env.WithCheckpointing(checkpointManager)
.RunStreamingAsync(workflow, "Hello"))
{
await foreach (WorkflowEvent evt in firstRun.WatchStreamAsync(blockOnPendingRequest: false))
{
if (evt is SuperStepCompletedEvent step && step.CompletionInfo?.Checkpoint is { } cp)
{
checkpoint = cp;
}
}
checkpoint.Should().NotBeNull();
}
// Act: Resume from the checkpoint and consume events so the run loop processes.
await using StreamingRun resumed = await env.WithCheckpointing(checkpointManager)
@@ -159,29 +158,29 @@ public class CheckpointResumeTests
InProcessExecutionEnvironment env = environment.ToWorkflowExecutionEnvironment();
// First run: collect checkpoint + pending request.
await using StreamingRun firstRun = await env.WithCheckpointing(checkpointManager)
.RunStreamingAsync(workflow, "Hello");
ExternalRequest? pendingRequest = null;
CheckpointInfo? checkpoint = null;
await foreach (WorkflowEvent evt in firstRun.WatchStreamAsync(blockOnPendingRequest: false))
await using (StreamingRun firstRun = await env.WithCheckpointing(checkpointManager)
.RunStreamingAsync(workflow, "Hello"))
{
if (evt is RequestInfoEvent requestInfo)
await foreach (WorkflowEvent evt in firstRun.WatchStreamAsync(blockOnPendingRequest: false))
{
pendingRequest = requestInfo.Request;
if (evt is RequestInfoEvent requestInfo)
{
pendingRequest = requestInfo.Request;
}
if (evt is SuperStepCompletedEvent step && step.CompletionInfo?.Checkpoint is { } cp)
{
checkpoint = cp;
}
}
if (evt is SuperStepCompletedEvent step && step.CompletionInfo?.Checkpoint is { } cp)
{
checkpoint = cp;
}
pendingRequest.Should().NotBeNull();
checkpoint.Should().NotBeNull();
}
pendingRequest.Should().NotBeNull();
checkpoint.Should().NotBeNull();
await firstRun.DisposeAsync();
// Act: Resume and respond to the restored request.
await using StreamingRun resumed = await env.WithCheckpointing(checkpointManager)
.ResumeStreamingAsync(workflow, checkpoint!);
@@ -293,11 +292,14 @@ public class CheckpointResumeTests
CheckpointManager checkpointManager = CheckpointManager.CreateInMemory();
InProcessExecutionEnvironment env = environment.ToWorkflowExecutionEnvironment();
await using StreamingRun firstRun = await env.WithCheckpointing(checkpointManager)
.RunStreamingAsync(workflow, "Hello");
ExternalRequest pendingRequest;
CheckpointInfo checkpoint;
(ExternalRequest pendingRequest, CheckpointInfo checkpoint) = await CapturePendingRequestAndCheckpointAsync(firstRun);
await firstRun.DisposeAsync();
await using (StreamingRun firstRun = await env.WithCheckpointing(checkpointManager)
.RunStreamingAsync(workflow, "Hello"))
{
(pendingRequest, checkpoint) = await CapturePendingRequestAndCheckpointAsync(firstRun);
}
// Act
await using StreamingRun resumed = await env.WithCheckpointing(checkpointManager)
@@ -346,20 +348,21 @@ public class CheckpointResumeTests
InProcessExecutionEnvironment env = environment.ToWorkflowExecutionEnvironment();
// First run: collect a checkpoint with pending requests.
await using StreamingRun firstRun = await env.WithCheckpointing(checkpointManager)
.RunStreamingAsync(workflow, "Hello");
CheckpointInfo? checkpoint = null;
await foreach (WorkflowEvent evt in firstRun.WatchStreamAsync(blockOnPendingRequest: false))
{
if (evt is SuperStepCompletedEvent step && step.CompletionInfo?.Checkpoint is { } cp)
{
checkpoint = cp;
}
}
checkpoint.Should().NotBeNull();
await firstRun.DisposeAsync();
await using (StreamingRun firstRun = await env.WithCheckpointing(checkpointManager)
.RunStreamingAsync(workflow, "Hello"))
{
await foreach (WorkflowEvent evt in firstRun.WatchStreamAsync(blockOnPendingRequest: false))
{
if (evt is SuperStepCompletedEvent step && step.CompletionInfo?.Checkpoint is { } cp)
{
checkpoint = cp;
}
}
checkpoint.Should().NotBeNull();
}
// Act: Resume with republishPendingEvents: false via the internal API.
await using StreamingRun resumed = await env.WithCheckpointing(checkpointManager)