Fix for terminations not completing as expected

src/Dapr.Workflow/Worker/WorkflowWorker.cs

@@ -21,7 +21,6 @@
 using Dapr.Workflow.Versioning;
 using Dapr.Workflow.Worker.Grpc;
 using Dapr.Workflow.Worker.Internal;
-using Grpc.Core;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging;
@@ -109,6 +108,29 @@ private async Task<OrchestratorResponse> HandleOrchestratorResponseAsync(Orchest
                 }
             }
 
+            //If the most recent event is `ExecutionTerminated`, acknowledge termination immediately
+            var timelineEvents = allPastEvents.Concat(request.NewEvents).ToList();
+            var latestEvent = timelineEvents.Count > 0 ? timelineEvents[^1] : null;
+
+            if (latestEvent?.ExecutionTerminated != null)
+            {
+                return new OrchestratorResponse
+                {
+                    InstanceId = request.InstanceId,
+                    CompletionToken = completionToken,
+                    Actions =
+                    {
+                        new OrchestratorAction
+                        {
+                            CompleteOrchestration = new CompleteOrchestrationAction
+                            {
+                                OrchestrationStatus = OrchestrationStatus.Terminated
+                            }
+                        }
+                    }
+                };
+            }
+
             // Create a new version tracker for this turn
             var versionTracker = new WorkflowVersionTracker(allPastEvents);
 

test/Dapr.Workflow.Test/Worker/WorkflowWorkerTests.cs

@@ -160,6 +160,99 @@ public async Task ExecuteAsync_ShouldComplete_WhenGrpcStreamCompletesImmediately
 
         await executeTask;
     }
+
+    [Fact]
+    public async Task HandleOrchestratorResponseAsync_ShouldReturnTerminatedCompletion_WhenReplayLatestEventIsExecutionTerminated()
+    {
+        var sp = new ServiceCollection().BuildServiceProvider();
+        var serializer = new JsonWorkflowSerializer(new JsonSerializerOptions(JsonSerializerDefaults.Web));
+        var options = new WorkflowRuntimeOptions();
+
+        // Intentionally no workflow registrations: this verifies the termination path
+        // is acknowledged before workflow lookup/instantiation.
+        var worker = new WorkflowWorker(
+            CreateGrpcClientMock().Object,
+            new StubWorkflowsFactory(),
+            NullLoggerFactory.Instance,
+            serializer,
+            sp,
+            options);
+
+        var request = new OrchestratorRequest
+        {
+            InstanceId = "i",
+            PastEvents =
+            {
+                new HistoryEvent
+                {
+                    ExecutionStarted = new ExecutionStartedEvent { Name = "wf-not-registered", Input = "" }
+                }
+            },
+            NewEvents =
+            {
+                new HistoryEvent
+                {
+                    ExecutionTerminated = new ExecutionTerminatedEvent()
+                }
+            }
+        };
+
+        var response = await InvokeHandleOrchestratorResponseAsync(worker, request);
+
+        Assert.Equal("i", response.InstanceId);
+        var action = Assert.Single(response.Actions);
+        Assert.NotNull(action.CompleteOrchestration);
+        Assert.Equal(OrchestrationStatus.Terminated, action.CompleteOrchestration!.OrchestrationStatus);
+    }
+
+    [Fact]
+    public async Task HandleOrchestratorResponseAsync_ShouldNotReturnTerminatedCompletion_WhenReplayLatestEventIsNotExecutionTerminated()
+    {
+        var sp = new ServiceCollection().BuildServiceProvider();
+        var serializer = new JsonWorkflowSerializer(new JsonSerializerOptions(JsonSerializerDefaults.Web));
+        var options = new WorkflowRuntimeOptions();
+
+        // Intentionally no workflow registrations. If the termination short-circuit does NOT trigger,
+        // normal path should fail with WorkflowNotFound-style completion.
+        var worker = new WorkflowWorker(
+            CreateGrpcClientMock().Object,
+            new StubWorkflowsFactory(),
+            NullLoggerFactory.Instance,
+            serializer,
+            sp,
+            options);
+
+        var request = new OrchestratorRequest
+        {
+            InstanceId = "i",
+            PastEvents =
+            {
+                new HistoryEvent
+                {
+                    ExecutionStarted = new ExecutionStartedEvent { Name = "wf-not-registered", Input = "" }
+                }
+            },
+            NewEvents =
+            {
+                new HistoryEvent
+                {
+                    ExecutionTerminated = new ExecutionTerminatedEvent()
+                },
+                new HistoryEvent
+                {
+                    OrchestratorStarted = new OrchestratorStartedEvent()
+                }
+            }
+        };
+
+        var response = await InvokeHandleOrchestratorResponseAsync(worker, request);
+
+        Assert.Equal("i", response.InstanceId);
+        var action = Assert.Single(response.Actions);
+        Assert.NotNull(action.CompleteOrchestration);
+        Assert.NotEqual(OrchestrationStatus.Terminated, action.CompleteOrchestration!.OrchestrationStatus);
+        Assert.Equal(OrchestrationStatus.Failed, action.CompleteOrchestration.OrchestrationStatus);
+    }
 
     [Fact]
     public async Task ExecuteAsync_ShouldSwallowOperationCanceledException_WhenStoppingTokenIsCanceled()