fix(worker): retain pending releases for unexpected terminal failures

Reuse the existing failed-result conversion after the full pre-close snapshot; preserve historical normal timeout actions and original failure details.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Copilot-Session: 974213a5-9101-41b4-9bde-aa86a72f25e1

Author: andystaples

durabletask/worker.py

@@ -2396,6 +2396,7 @@ def execute(
         )
         result: ExecutionResults | None = None
         dispatch_exited = False
+        unexpected_failure = False
         try:
             try:
                 # Rebuild local state by replaying old history into the orchestrator function
@@ -2437,6 +2438,7 @@ def execute(
                     f"{instance_id}: Orchestrator {orchestration_name} exited unexpectedly.",
                     exc_info=True,
                 )
+                unexpected_failure = not ctx._is_complete  # pyright: ignore[reportPrivateUsage]
                 ctx.set_failed(ex)
 
             if not ctx._is_complete:  # pyright: ignore[reportPrivateUsage]
@@ -2467,6 +2469,13 @@ def execute(
                 # Snapshot before generator teardown can exit a critical section.
                 entity_lock_releases=list(ctx._entity_context.get_lock_release_messages()),  # pyright: ignore[reportPrivateUsage]
             )
+            if unexpected_failure and result.entity_lock_releases:
+                failure_details = next(
+                    action.completeOrchestration.failureDetails for action in result.actions
+                    if action.HasField("completeOrchestration")
+                    and action.completeOrchestration.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED
+                )
+                result = result.as_failed(failure_details)
         except BaseException:
             dispatch_exited = True
             raise

tests/durabletask/test_orchestration_dispatch_recovery.py

@@ -190,6 +190,54 @@ def orchestrator(ctx: task.OrchestrationContext, _: Any) -> Generator[task.Task[
     assert completion.failureDetails.errorMessage == str(error)
 
 
+@pytest.mark.parametrize("fail_during_teardown", [False, True])
+def test_terminal_failure_after_pending_call_timeout_preserves_missing_release(fail_during_teardown: bool):
+    entity_a = EntityInstanceId("Counter", "a")
+    entity_b = EntityInstanceId("Counter", "b")
+    error = TeardownExit("terminal failure after pending call timeout")
+
+    def orchestrator(ctx: task.OrchestrationContext, _: Any) -> Generator[task.Task[Any], Any, Any]:
+        after_timeout = False
+        try:
+            with (yield ctx.lock_entities([entity_a, entity_b])):
+                call = ctx.call_entity(entity_a, "get")
+                yield task.when_any([call, ctx.wait_for_external_event("timeout")])
+            after_timeout = True
+            yield ctx.call_activity("after_timeout")
+            if fail_during_teardown:
+                yield ctx.wait_for_external_event("never")
+            return _FailingOutput(error)
+        finally:
+            if fail_during_teardown and after_timeout:
+                raise error
+
+    executor = _executor(orchestrator)
+    started = list(_request().newEvents)
+    initial = executor.execute(INSTANCE_ID, [], started)
+    lock_id = initial.actions[0].sendEntityMessage.entityLockRequested.criticalSectionId
+    granted = [
+        helpers.new_event_sent_event(1, str(entity_a), json.dumps({"id": lock_id})),
+        helpers.new_event_raised_event(lock_id, None),
+    ]
+    calling = executor.execute(INSTANCE_ID, started, granted)
+    call_id = calling.actions[0].sendEntityMessage.entityOperationCalled.requestId
+    history = started + granted + [
+        helpers.new_event_sent_event(2, str(entity_a), json.dumps({"id": call_id})),
+        helpers.new_event_raised_event("timeout", None),
+        helpers.new_event_sent_event(3, str(entity_b), None),
+        helpers.new_task_scheduled_event(4, "after_timeout"),
+    ]
+    result = executor.execute(INSTANCE_ID, history, [helpers.new_task_completed_event(4, '"done"')])
+
+    assert len(result.actions) == 2
+    release = result.actions[0].sendEntityMessage.entityUnlockSent
+    assert release.targetInstanceId.value == str(entity_a)
+    assert release.criticalSectionId == lock_id
+    completion = result.actions[1].completeOrchestration
+    assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED
+    assert completion.failureDetails.errorMessage == str(error)
+
+
 @pytest.mark.parametrize("during_serialization", [False, True])
 def test_unexpected_exit_commits_failure_and_allows_next_dispatch(during_serialization: bool):
     def orchestrator(ctx: task.OrchestrationContext, _: Any) -> Generator[task.Task[Any], Any, None]: