From 87a8fa2a9d0ffd3d0b4b31882e70c4f60462bb29 Mon Sep 17 00:00:00 2001
From: Peter Ibekwe <109177538+peibekwe@users.noreply.github.com>
Date: Wed, 15 Apr 2026 21:20:45 -0700
Subject: [PATCH] .NET: Fix intermittent checkpoint-restore race in in-process
 workflow runs (#5134)

* Improve workflow unit tests

* Update test name prefix for clarity.

* Update tests to surface any errors.

* fix check-point restore-time race in off-thread workflow event stream

* Fixes an intermittent checkpoint-restore race in in-process workflow runs.
---
 .../InProc/InProcessRunnerContext.cs          |  6 +++
 .../CheckpointResumeTests.cs                  | 42 +++++++++++++++++++
 2 files changed, 48 insertions(+)

diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs
index f0bb8cac26..d6c7d301e3 100644
--- a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs
+++ b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs
@@ -419,6 +419,12 @@ internal sealed class InProcessRunnerContext : IRunnerContext
                                                       .Select(id => this.EnsureExecutorAsync(id, tracer: null).AsTask())
                                                       .ToArray();
 
+        // Discard queued external deliveries from the superseded timeline so a runtime
+        // restore cannot apply stale responses after importing the checkpoint state.
+        while (this._queuedExternalDeliveries.TryDequeue(out _))
+        {
+        }
+
         this._nextStep = new StepContext();
         this._nextStep.ImportMessages(importedState.QueuedMessages);
 
diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/CheckpointResumeTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/CheckpointResumeTests.cs
index 9d4b514af7..53ea644712 100644
--- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/CheckpointResumeTests.cs
+++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/CheckpointResumeTests.cs
@@ -279,6 +279,48 @@ public class CheckpointResumeTests
             "the workflow should be able to continue after the runtime restore replay");
     }
 
+    /// <summary>
+    /// Verifies that restoring a live run clears any queued external responses from the
+    /// superseded timeline before importing checkpoint state.
+    /// </summary>
+    [Fact]
+    internal async Task Checkpoint_Restore_ClearsQueuedExternalResponsesBeforeImportAsync()
+    {
+        Workflow workflow = CreateSimpleRequestWorkflow();
+        CheckpointManager checkpointManager = CheckpointManager.CreateInMemory();
+        InProcessExecutionEnvironment env = ExecutionEnvironment.InProcess_Lockstep.ToWorkflowExecutionEnvironment();
+
+        await using StreamingRun run = await env.WithCheckpointing(checkpointManager)
+                                                .RunStreamingAsync(workflow, "Hello");
+
+        (ExternalRequest pendingRequest, CheckpointInfo checkpoint) = await CapturePendingRequestAndCheckpointAsync(run);
+
+        await run.SendResponseAsync(pendingRequest.CreateResponse("World"));
+        await run.RestoreCheckpointAsync(checkpoint);
+
+        List<WorkflowEvent> restoredEvents = await ReadToHaltAsync(run);
+        ExternalRequest replayedRequest = restoredEvents.OfType<RequestInfoEvent>()
+                                                        .Select(evt => evt.Request)
+                                                        .Should()
+                                                        .ContainSingle("the restored run should still be waiting for the checkpointed request")
+                                                        .Subject;
+
+        restoredEvents.OfType<WorkflowErrorEvent>().Should().BeEmpty(
+            "a queued response from the superseded timeline should not be processed after restore");
+        RunStatus statusAfterRestore = await run.GetStatusAsync();
+        statusAfterRestore.Should().Be(RunStatus.PendingRequests,
+            "the restored run should remain pending until a post-restore response is sent");
+
+        await run.SendResponseAsync(replayedRequest.CreateResponse("Again"));
+
+        List<WorkflowEvent> completionEvents = await ReadToHaltAsync(run);
+        completionEvents.OfType<WorkflowErrorEvent>().Should().BeEmpty(
+            "the restored request should complete cleanly once a new response is provided");
+        RunStatus finalStatus = await run.GetStatusAsync();
+        finalStatus.Should().Be(RunStatus.Idle,
+            "the workflow should finish once the replayed request receives a fresh response");
+    }
+
     /// <summary>
     /// Verifies that a resumed parent workflow re-emits pending requests that originated in a subworkflow.
     /// </summary>