Modernize newly merged code for Python 3.11

Use native exception-group matching and preserve nested failure diagnostics and notes. Extend Self typing to immutable entity copy protocols, remove the obsolete asyncio backport dependency, and cover dispatch and provider compatibility.

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

Copilot-Session: e6b07f3f-b3e1-49dd-9121-318968272d31

Author: Bernd Verst

CHANGELOG.md

@@ -15,14 +15,19 @@ by one or more task hubs.
 CHANGED
 
 - Dropped Python 3.10 support. Python 3.11 or later is now required.
+- Removed the obsolete PyPI `asyncio` backport dependency; all supported
+Python versions use the standard-library module.
 - Context managers now preserve subclass types in type checkers, including
 Azure Managed clients and workers.
+- Entity ID copy and pickle hooks now preserve subclass types in type checkers.
 - Multiple task completion callback failures now raise the built-in
 `ExceptionGroup`, preserving each exception and its traceback. A single callback
 failure is still raised directly.
 
 FIXED
 
+- Grouped failures now retain nested exception messages, tracebacks, and notes
+in failure details, including `ExceptionGroup` failures from task callbacks.
 - Implemented `__hash__` and `__repr__` for `EntityInstanceId`, making identity attributes immutable to guarantee hash stability when used in sets or dict keys.
 - Large-payload externalization now checks every eligible payload's maximum
 size before uploading or replacing any protobuf values, including carryover

azure-functions-durable/CHANGELOG.md

@@ -27,6 +27,8 @@ directly.
 
 FIXED
 
+- With the corresponding core SDK update, grouped failures retain nested
+exception messages, tracebacks, and notes in failure details.
 - With the corresponding core `durabletask` SDK fix, oversized messages fail
 large-payload validation before any uploads or payload replacements. Size
 errors now take precedence over hypothetical earlier storage failures.

durabletask-azuremanaged/CHANGELOG.md

@@ -18,6 +18,8 @@ directly.
 
 FIXED
 
+- With the corresponding core SDK update, grouped failures retain nested
+exception messages, tracebacks, and notes in failure details.
 - With the corresponding core `durabletask` SDK fix, oversized messages fail
 large-payload validation before any uploads or payload replacements. Size
 errors now take precedence over hypothetical earlier storage failures.

durabletask/entities/entity_instance_id.py

@@ -1,6 +1,8 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 
+from typing import Self
+
 
 class EntityInstanceId:
     __slots__ = ("_entity", "_key")
@@ -29,13 +31,13 @@ def __delattr__(self, name: str) -> None:
             f"'{type(self).__name__}' object does not support attribute deletion"
         )
 
-    def __copy__(self) -> "EntityInstanceId":
+    def __copy__(self) -> Self:
         return self
 
-    def __deepcopy__(self, memo: object) -> "EntityInstanceId":
+    def __deepcopy__(self, memo: object) -> Self:
         return self
 
-    def __reduce__(self) -> tuple[type["EntityInstanceId"], tuple[str, str]]:
+    def __reduce__(self) -> tuple[type[Self], tuple[str, str]]:
         return (type(self), (self._entity, self._key))
 
     def __str__(self) -> str:

durabletask/internal/helpers.py

@@ -258,8 +258,9 @@ def new_failure_details(
         errorType=get_qualified_name(type(ex)),
         errorMessage=str(ex),
         stackTrace=wrappers_pb2.StringValue(value=''.join(
-            traceback.format_tb(ex.__traceback__) if isinstance(ex, Exception)
-            else traceback.format_exception(type(ex), ex, ex.__traceback__))),
+            traceback.format_tb(ex.__traceback__)
+            if isinstance(ex, Exception) and not isinstance(ex, BaseExceptionGroup)
+            else traceback.format_exception(ex))),
         innerFailure=(
             new_failure_details(
                 inner, exception_properties_provider, logger, _visited)

durabletask/worker.py

@@ -7,7 +7,6 @@
 import json
 import logging
 import os
-import sys
 import time
 from collections.abc import Callable, Generator, Sequence
 from concurrent.futures import ThreadPoolExecutor
@@ -64,11 +63,11 @@
 
 
 def _is_control_flow_exception(error: BaseException) -> bool:
-    if isinstance(error, (asyncio.CancelledError, GeneratorExit, KeyboardInterrupt, SystemExit)):
-        return True
-    if sys.version_info >= (3, 11) and isinstance(error, BaseExceptionGroup):
-        return any(_is_control_flow_exception(inner) for inner in error.exceptions)
-    return False
+    control_flow_types = (asyncio.CancelledError, GeneratorExit, KeyboardInterrupt, SystemExit)
+    if isinstance(error, BaseExceptionGroup):
+        group = cast(BaseExceptionGroup[BaseException], error)
+        return group.subgroup(control_flow_types) is not None
+    return isinstance(error, control_flow_types)
 
 
 class ConcurrencyOptions:

pyproject.toml

@@ -32,7 +32,6 @@ readme = "README.md"
 dependencies = [
     "grpcio",
     "protobuf",
-    "asyncio",
     "packaging"
 ]
 

requirements.txt

@@ -8,7 +8,6 @@ pytest-cov
 azure-identity
 azure-functions
 typing-extensions
-asyncio
 packaging
 opentelemetry-api
 opentelemetry-sdk

tests/azure-functions-durable/test_worker_compat.py

@@ -84,6 +84,31 @@ def orchestrator(context):
     assert len(response.actions) == 1
 
 
+@pytest.mark.parametrize("use_async", [False, True])
+async def test_worker_preserves_grouped_failure_diagnostics(monkeypatch, use_async):
+    monkeypatch.setattr(payloads, "_payload_store", None)
+
+    def orchestrator(context):
+        error = ValueError("nested failure")
+        error.add_note("nested note")
+        raise ExceptionGroup("activity group", [error, RuntimeError("other failure")])
+
+    worker = DurableFunctionsWorker()
+    request = _encode_orchestrator_request("grouped-failure")
+    if use_async:
+        encoded = await worker.execute_orchestration_request_async(orchestrator, request)
+    else:
+        encoded = worker.execute_orchestration_request(orchestrator, request)
+    response = _decode_orchestrator_response(encoded)
+    completion = _get_completion_action(response)
+
+    assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED
+    assert completion.failureDetails.errorType == "builtins.ExceptionGroup"
+    assert "ValueError: nested failure" in completion.failureDetails.stackTrace.value
+    assert "RuntimeError: other failure" in completion.failureDetails.stackTrace.value
+    assert "nested note" in completion.failureDetails.stackTrace.value
+
+
 def test_worker_created_before_configuration_hydrates_and_externalizes(monkeypatch, payload_store_factory):
     monkeypatch.setattr(payloads, "_payload_store", None)
     worker = DurableFunctionsWorker()

tests/durabletask-azuremanaged/test_azuremanaged_grpc_resiliency.py

@@ -71,6 +71,32 @@ def orchestrator(ctx: task.OrchestrationContext, _: Any):
     assert response.actions[0].completeOrchestration.failureDetails.errorMessage == "deterministic teardown failure"
 
 
+def test_dts_worker_preserves_grouped_failure_diagnostics():
+    def orchestrator(ctx: task.OrchestrationContext, _: Any):
+        error = ValueError("nested failure")
+        error.add_note("nested note")
+        raise ExceptionGroup("activity group", [error, RuntimeError("other failure")])
+
+    with DurableTaskSchedulerWorker(
+            host_address="localhost:4001", taskhub="hub", token_credential=None) as worker:
+        name = worker.add_orchestrator(orchestrator)
+        request = pb.OrchestratorRequest(instanceId="instance", newEvents=[
+            helpers.new_orchestrator_started_event(),
+            helpers.new_execution_started_event(name, "instance"),
+        ])
+        stub = Mock()
+        worker._execute_orchestrator(request, stub, "original-token")
+
+    stub.CompleteOrchestratorTask.assert_called_once()
+    stub.AbandonTaskOrchestratorWorkItem.assert_not_called()
+    completion = stub.CompleteOrchestratorTask.call_args.args[0].actions[0].completeOrchestration
+    assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED
+    assert completion.failureDetails.errorType == "builtins.ExceptionGroup"
+    assert "ValueError: nested failure" in completion.failureDetails.stackTrace.value
+    assert "RuntimeError: other failure" in completion.failureDetails.stackTrace.value
+    assert "nested note" in completion.failureDetails.stackTrace.value
+
+
 def test_dts_worker_passes_exception_properties_provider_to_base_worker():
     class PropertiesProvider:
         def get_exception_properties(self, exception: Exception) -> Mapping[str, Any] | None: