Merge branch 'main' into wangbill/serverless-op1

Author: YunchuWang

.github/workflows/typecheck.yml

@@ -0,0 +1,38 @@
+name: Type Check (pyright)
+
+on:
+  push:
+    branches:
+      - "main"
+    tags:
+      - "v*"
+      - "azuremanaged-v*"
+  pull_request:
+    branches:
+      - "main"
+
+permissions:
+  contents: read
+
+jobs:
+  pyright:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Set up Python 3.10 (lowest supported)
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.10"
+
+      - name: Install packages and dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r requirements.txt
+          pip install -e ".[azure-blob-payloads,opentelemetry]"
+          pip install -e ./durabletask-azuremanaged
+          pip install pyright
+
+      - name: Run pyright (strict, Python 3.10)
+        run: pyright

.gitignore

@@ -131,5 +131,7 @@ dmypy.json
 
 # IDEs
 .idea
+.worktrees/
+docs/superpowers/
 
-coverage.lcov
+coverage.lcov

CHANGELOG.md

@@ -5,12 +5,47 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project
 adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## Unreleased
-
 ## v1.5.0
 
+BREAKING CHANGES (type-level only — no runtime impact for typical users)
+
+These changes do not alter runtime behavior for clients or activity/orchestrator
+authors, but because the package ships `py.typed`, consumers running strict type
+checkers (pyright/mypy) against their own code — or subclassing the public
+abstract types — may see new type-check errors and need to update their
+overrides:
+
+- `OrchestrationContext.create_timer` now returns the specific `TimerTask` type
+  (was `CancellableTask`)
+  ([#93](https://github.com/microsoft/durabletask-python/issues/93)).
+- `OrchestrationContext.wait_for_external_event` now returns `CancellableTask[Any]`
+  (was a bare `CancellableTask`).
+- `WhenAnyTask` is now generic; `when_any(tasks: Sequence[Task[T]])` returns
+  `WhenAnyTask[T]` for better static inference of the completing child task
+  ([#94](https://github.com/microsoft/durabletask-python/issues/94)).
+  `CompositeTask.on_child_completed` now takes `Task[Any]`.
+- `TaskHubGrpcWorker.add_activity` / `add_entity` (and the internal registry
+  methods) now require `Activity[Any, Any]` / `Entity[Any, Any]` instead of the
+  bare `Activity` / `Entity` aliases.
+- `OrchestrationContext.call_entity` / `signal_entity` `input` parameter widened
+  from `TInput | None` to `Any` (Liskov-safe for callers; subclass overrides
+  using the old narrower type will be flagged).
+- gRPC client interceptors now use the public `grpc.ClientCallDetails` /
+  `grpc.aio.ClientCallDetails` types instead of private internal namedtuples;
+  custom interceptor subclasses should retype their override parameters.
+- These changes also broadly improve generic type-safety hints throughout the
+  SDK ([#92](https://github.com/microsoft/durabletask-python/issues/92)).
+
 ADDED
 
+- Added context-manager support (`__enter__` / `__exit__`) to
+  `TaskHubGrpcClient` so it can be used with `with` statements, mirroring the
+  existing `AsyncTaskHubGrpcClient` async-context-manager support and the
+  `TaskHubGrpcWorker` pattern. `DurableTaskSchedulerClient` inherits this
+  behavior automatically. `__exit__` delegates to `close()`, so the
+  resiliency-aware teardown (in-flight recreate thread join, retired-channel
+  timer cancellation, and SDK-owned channel cleanup) runs unchanged through the
+  new `with` path.
 - Added `ReplaySafeLogger` and `OrchestrationContext.create_replay_safe_logger()`
   for suppressing duplicate log messages during orchestrator replay
 - Added `GrpcChannelOptions` and `GrpcRetryPolicyOptions` for configuring
@@ -20,8 +55,41 @@ ADDED
   `TaskHubGrpcClient`, `AsyncTaskHubGrpcClient`, and `TaskHubGrpcWorker` to
   support pre-configured channel passthrough and low-level gRPC channel
   customization.
-- Added `get_orchestration_history()` and `list_instance_ids()` to the sync and async gRPC clients.
-- Added in-memory backend support for `StreamInstanceHistory` and `ListInstanceIds` so local orchestration tests can retrieve history and page terminal instance IDs by completion window.
+- Added `GrpcWorkerResiliencyOptions` and `GrpcClientResiliencyOptions`, plus
+  `resiliency_options` constructor parameters on `TaskHubGrpcClient`,
+  `AsyncTaskHubGrpcClient`, and `TaskHubGrpcWorker`, to configure hello
+  deadlines, silent-disconnect detection, reconnect backoff, and channel
+  recreation thresholds for SDK-managed gRPC connections.
+- Added `get_orchestration_history()` and `list_instance_ids()` to the sync
+  and async gRPC clients.
+- Added in-memory backend support for `StreamInstanceHistory` and
+  `ListInstanceIds` so local orchestration tests can retrieve history and page
+  terminal instance IDs by completion window.
+
+CHANGED
+
+- `when_any` now copies its input into a new list (`WhenAnyTask(list(tasks))`).
+  Previously the task aliased the caller's list, so mutating it after
+  construction was visible inside the task; that side effect no longer occurs.
+
+FIXED
+
+- Fixed `EntityInstanceId.__lt__` infinite recursion when compared against a
+  non-`EntityInstanceId` operand. It now returns `NotImplemented`, so mixed-type
+  comparisons raise `TypeError` cleanly instead of recursing.
+- Improved `TaskHubGrpcWorker` recovery from stale or disconnected gRPC streams
+  so configured hello timeouts apply on fresh connections, received work resets
+  failure tracking, SDK-owned channels are refreshed and cleaned up safely, and
+  caller-owned channels are never recreated or closed during reconnects.
+- Fixed `TaskHubGrpcWorker` so in-flight and queued work item completions keep
+  draining across graceful gRPC stream resets and worker shutdown before the
+  worker retires an SDK-owned channel.
+- Improved sync and async gRPC clients so repeated transport failures recreate
+  SDK-owned channels, while long-poll deadlines, successful replies, and
+  application-level RPC errors do not trigger unnecessary channel replacement.
+- Fixed `TaskHubGrpcClient.close()` so explicit sync client shutdown now closes
+  any previously retired SDK-owned gRPC channels immediately instead of waiting
+  for the delayed cleanup timer.
 
 ## v1.4.0
 

dev-requirements.txt

@@ -1,2 +1,3 @@
 grpcio-tools
 pymarkdownlnt
+pyright

durabletask-azuremanaged/CHANGELOG.md

@@ -14,10 +14,25 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 ## v1.5.0
 
 - Updates base dependency to durabletask v1.5.0
+- Improved type coverage benefits Azure Managed users: `create_timer` now
+  returns the specific `TimerTask` type and `when_any` is generic so the
+  completing child task is type-checked through `DurableTaskSchedulerClient`,
+  `AsyncDurableTaskSchedulerClient`, and `DurableTaskSchedulerWorker` derived
+  orchestrations.
+- gRPC client interceptors in the core SDK now use the public
+  `grpc.ClientCallDetails` / `grpc.aio.ClientCallDetails` types instead of
+  private internal namedtuples. Any custom DTS auth interceptor built on the
+  same pattern as `DTSDefaultClientInterceptorImpl` should retype its
+  `_intercept_call` override parameter accordingly. This is a type-level change
+  only and does not alter runtime behavior.
 - Added optional `interceptors`, `channel`, and `channel_options` parameters to
   `DurableTaskSchedulerClient`, `AsyncDurableTaskSchedulerClient`, and
   `DurableTaskSchedulerWorker` to allow combining custom gRPC interceptors with
   DTS defaults and to support pre-configured/customized gRPC channels.
+- Added pass-through `resiliency_options` support on
+  `DurableTaskSchedulerClient`, `AsyncDurableTaskSchedulerClient`, and
+  `DurableTaskSchedulerWorker` so Azure Managed applications can use the core
+  SDK's gRPC resiliency option types through their constructors.
 - Added `workerid` gRPC metadata on Durable Task Scheduler worker calls for
   improved worker identity and observability.
 - Improved sync access token refresh concurrency handling to avoid duplicate

durabletask-azuremanaged/durabletask/azuremanaged/client.py

@@ -2,8 +2,7 @@
 # Licensed under the MIT License.
 
 import logging
-
-from typing import Optional, Sequence
+from collections.abc import Sequence
 
 import grpc
 import grpc.aio
@@ -15,7 +14,10 @@
     DTSDefaultClientInterceptorImpl,
 )
 from durabletask.client import AsyncTaskHubGrpcClient, TaskHubGrpcClient
-from durabletask.grpc_options import GrpcChannelOptions
+from durabletask.grpc_options import (
+    GrpcChannelOptions,
+    GrpcClientResiliencyOptions,
+)
 import durabletask.internal.shared as shared
 from durabletask.payload.store import PayloadStore
 
@@ -25,15 +27,16 @@ class DurableTaskSchedulerClient(TaskHubGrpcClient):
     def __init__(self, *,
                  host_address: str,
                  taskhub: str,
-                 token_credential: Optional[TokenCredential],
-                 channel: Optional[grpc.Channel] = None,
+                 token_credential: TokenCredential | None,
+                 channel: grpc.Channel | None = None,
                  secure_channel: bool = True,
-                 interceptors: Optional[Sequence[shared.ClientInterceptor]] = None,
-                 channel_options: Optional[GrpcChannelOptions] = None,
-                 default_version: Optional[str] = None,
-                 payload_store: Optional[PayloadStore] = None,
-                 log_handler: Optional[logging.Handler] = None,
-                 log_formatter: Optional[logging.Formatter] = None):
+                 interceptors: Sequence[shared.ClientInterceptor] | None = None,
+                 channel_options: GrpcChannelOptions | None = None,
+                 resiliency_options: GrpcClientResiliencyOptions | None = None,
+                 default_version: str | None = None,
+                 payload_store: PayloadStore | None = None,
+                 log_handler: logging.Handler | None = None,
+                 log_formatter: logging.Formatter | None = None):
 
         if not taskhub:
             raise ValueError("Taskhub value cannot be empty. Please provide a value for your taskhub")
@@ -54,6 +57,7 @@ def __init__(self, *,
             log_formatter=log_formatter,
             interceptors=resolved_interceptors,
             channel_options=channel_options,
+            resiliency_options=resiliency_options,
             default_version=default_version,
             payload_store=payload_store)
 
@@ -70,15 +74,17 @@ class AsyncDurableTaskSchedulerClient(AsyncTaskHubGrpcClient):
     Args:
         host_address (str): The gRPC endpoint address of the DTS service.
         taskhub (str): The name of the task hub. Cannot be empty.
-        token_credential (Optional[TokenCredential]): Azure credential for authentication.
+        token_credential (TokenCredential | None): Azure credential for authentication.
             If None, anonymous authentication will be used.
         secure_channel (bool, optional): Whether to use a secure gRPC channel (TLS).
             Defaults to True.
-        default_version (Optional[str], optional): Default version string for orchestrations.
-        payload_store (Optional[PayloadStore], optional): A payload store for
+        resiliency_options (GrpcClientResiliencyOptions | None, optional): Client-side
+            gRPC resiliency settings forwarded to the base async client.
+        default_version (str | None, optional): Default version string for orchestrations.
+        payload_store (PayloadStore | None, optional): A payload store for
             externalizing large payloads. If None, payloads are sent inline.
-        log_handler (Optional[logging.Handler], optional): Custom logging handler for client logs.
-        log_formatter (Optional[logging.Formatter], optional): Custom log formatter for client logs.
+        log_handler (logging.Handler | None, optional): Custom logging handler for client logs.
+        log_formatter (logging.Formatter | None, optional): Custom log formatter for client logs.
 
     Raises:
         ValueError: If taskhub is empty or None.
@@ -99,15 +105,16 @@ class AsyncDurableTaskSchedulerClient(AsyncTaskHubGrpcClient):
     def __init__(self, *,
                  host_address: str,
                  taskhub: str,
-                 token_credential: Optional[AsyncTokenCredential],
-                 channel: Optional[grpc.aio.Channel] = None,
+                 token_credential: AsyncTokenCredential | None,
+                 channel: grpc.aio.Channel | None = None,
                  secure_channel: bool = True,
-                 interceptors: Optional[Sequence[shared.AsyncClientInterceptor]] = None,
-                 channel_options: Optional[GrpcChannelOptions] = None,
-                 default_version: Optional[str] = None,
-                 payload_store: Optional[PayloadStore] = None,
-                 log_handler: Optional[logging.Handler] = None,
-                 log_formatter: Optional[logging.Formatter] = None):
+                 interceptors: Sequence[shared.AsyncClientInterceptor] | None = None,
+                 channel_options: GrpcChannelOptions | None = None,
+                 resiliency_options: GrpcClientResiliencyOptions | None = None,
+                 default_version: str | None = None,
+                 payload_store: PayloadStore | None = None,
+                 log_handler: logging.Handler | None = None,
+                 log_formatter: logging.Formatter | None = None):
 
         if not taskhub:
             raise ValueError("Taskhub value cannot be empty. Please provide a value for your taskhub")
@@ -128,5 +135,6 @@ def __init__(self, *,
             log_formatter=log_formatter,
             interceptors=resolved_interceptors,
             channel_options=channel_options,
+            resiliency_options=resiliency_options,
             default_version=default_version,
             payload_store=payload_store)

durabletask-azuremanaged/durabletask/azuremanaged/internal/access_token_manager.py

@@ -2,7 +2,6 @@
 # Licensed under the MIT License.
 from datetime import datetime, timedelta, timezone
 from threading import Lock
-from typing import Optional
 
 from azure.core.credentials import AccessToken, TokenCredential
 from azure.core.credentials_async import AsyncTokenCredential
@@ -13,9 +12,9 @@
 # By default, when there's 10minutes left before the token expires, refresh the token
 class AccessTokenManager:
 
-    _token: Optional[AccessToken]
+    _token: AccessToken | None
 
-    def __init__(self, token_credential: Optional[TokenCredential], refresh_interval_seconds: int = 600):
+    def __init__(self, token_credential: TokenCredential | None, refresh_interval_seconds: int = 600):
         self._scope = "https://durabletask.io/.default"
         self._refresh_interval_seconds = refresh_interval_seconds
         self._logger = shared.get_logger("token_manager")
@@ -30,7 +29,7 @@ def __init__(self, token_credential: Optional[TokenCredential], refresh_interval
             self._token = None
             self.expiry_time = None
 
-    def get_access_token(self) -> Optional[AccessToken]:
+    def get_access_token(self) -> AccessToken | None:
         if self._token is None or self.is_token_expired():
             with self._refresh_lock:
                 if self._token is None or self.is_token_expired():
@@ -59,9 +58,9 @@ class AsyncAccessTokenManager:
 
     This avoids blocking the event loop when acquiring or refreshing tokens."""
 
-    _token: Optional[AccessToken]
+    _token: AccessToken | None
 
-    def __init__(self, token_credential: Optional[AsyncTokenCredential],
+    def __init__(self, token_credential: AsyncTokenCredential | None,
                  refresh_interval_seconds: int = 600):
         self._scope = "https://durabletask.io/.default"
         self._refresh_interval_seconds = refresh_interval_seconds
@@ -71,7 +70,7 @@ def __init__(self, token_credential: Optional[AsyncTokenCredential],
         self._token = None
         self.expiry_time = None
 
-    async def get_access_token(self) -> Optional[AccessToken]:
+    async def get_access_token(self) -> AccessToken | None:
         if self._token is None or self.is_token_expired():
             await self.refresh_token()
         return self._token

durabletask-azuremanaged/durabletask/azuremanaged/internal/durabletask_grpc_interceptor.py

@@ -2,7 +2,6 @@
 # Licensed under the MIT License.
 
 from importlib.metadata import version
-from typing import Optional
 
 import grpc
 from azure.core.credentials import TokenCredential
@@ -15,8 +14,6 @@
 from durabletask.internal.grpc_interceptor import (
     DefaultAsyncClientInterceptorImpl,
     DefaultClientInterceptorImpl,
-    _AsyncClientCallDetails,
-    _ClientCallDetails,
 )
 
 
@@ -27,9 +24,9 @@ class DTSDefaultClientInterceptorImpl (DefaultClientInterceptorImpl):
 
     def __init__(
             self,
-            token_credential: Optional[TokenCredential],
+            token_credential: TokenCredential | None,
             taskhub_name: str,
-            worker_id: Optional[str] = None):
+            worker_id: str | None = None):
         try:
             # Get the version of the azuremanaged package
             sdk_version = version('durabletask-azuremanaged')
@@ -63,7 +60,7 @@ def _upsert_authorization_header(self, token: str) -> None:
             self._metadata.append(("authorization", f"Bearer {token}"))
 
     def _intercept_call(
-            self, client_call_details: _ClientCallDetails) -> grpc.ClientCallDetails:
+            self, client_call_details: grpc.ClientCallDetails) -> grpc.ClientCallDetails:
         """Internal intercept_call implementation which adds metadata to grpc metadata in the RPC
             call details."""
         # Refresh the auth token if a credential was provided. The call to
@@ -83,7 +80,7 @@ class DTSAsyncDefaultClientInterceptorImpl(DefaultAsyncClientInterceptorImpl):
     This class implements async gRPC interceptors to add DTS-specific headers
     (task hub name, user agent, and authentication token) to all async calls."""
 
-    def __init__(self, token_credential: Optional[AsyncTokenCredential], taskhub_name: str):
+    def __init__(self, token_credential: AsyncTokenCredential | None, taskhub_name: str):
         try:
             # Get the version of the azuremanaged package
             sdk_version = version('durabletask-azuremanaged')
@@ -115,7 +112,7 @@ def _upsert_authorization_header(self, token: str) -> None:
             self._metadata.append(("authorization", f"Bearer {token}"))
 
     async def _intercept_call(
-            self, client_call_details: _AsyncClientCallDetails) -> grpc.aio.ClientCallDetails:
+            self, client_call_details: grpc.aio.ClientCallDetails) -> grpc.aio.ClientCallDetails:
         """Internal intercept_call implementation which adds metadata to grpc metadata in the RPC
             call details."""
         # Refresh the auth token if a credential was provided. The call to