diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/LockstepRunEventStream.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/LockstepRunEventStream.cs index c4c14c00a7..cdd8cc7686 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/LockstepRunEventStream.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/LockstepRunEventStream.cs @@ -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. diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/CheckpointResumeTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/CheckpointResumeTests.cs index 29e68cef0b..9d4b514af7 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/CheckpointResumeTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/CheckpointResumeTests.cs @@ -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 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)