fix: address conflict with main

Signed-off-by: Samantha Coyle <sam@diagrid.io>

Author: sicoyle

.github/workflows/pr-validation.yml

@@ -47,6 +47,12 @@ jobs:
         go install github.com/dapr/durabletask-go@main        
         durabletask-go --port 4001 &
         tox -e py${{ matrix.python-version }}-e2e
+    - name: Run examples
+      run: |
+        pip install mechanical-markdown
+        cd examples
+        durabletask-go --port 4001 &
+        mm.py README.md
   publish:
     needs: build
     if: startswith(github.ref, 'refs/tags/v')

durabletask/aio/client.py

@@ -27,6 +27,7 @@
 # If `opentelemetry-instrumentation-grpc` is available, enable the gRPC client interceptor
 try:
     from opentelemetry.instrumentation.grpc import GrpcInstrumentorClient
+
     GrpcInstrumentorClient().instrument()
 except ImportError:
     pass

durabletask/client.py

@@ -24,10 +24,12 @@
 # If `opentelemetry-instrumentation-grpc` is available, enable the gRPC client interceptor
 try:
     from opentelemetry.instrumentation.grpc import GrpcInstrumentorClient
+
     GrpcInstrumentorClient().instrument()
 except ImportError:
     pass
 
+
 class OrchestrationStatus(Enum):
     """The status of an orchestration instance."""
 

durabletask/internal/helpers.py

@@ -204,14 +204,19 @@ def new_create_timer_action(id: int, fire_at: datetime) -> pb.OrchestratorAction
 
 
 def new_schedule_task_action(
-    id: int, name: str, encoded_input: Optional[str], router: Optional[pb.TaskRouter] = None
+    id: int,
+    name: str,
+    encoded_input: Optional[str],
+    router: Optional[pb.TaskRouter] = None,
+    task_execution_id: str = "",
 ) -> pb.OrchestratorAction:
     return pb.OrchestratorAction(
         id=id,
         scheduleTask=pb.ScheduleTaskAction(
             name=name,
             input=get_string_value(encoded_input),
             router=router,
+            taskExecutionId=task_execution_id,
         ),
         router=router,
     )

durabletask/internal/shared.py

@@ -50,6 +50,14 @@ def get_default_host_address() -> str:
     return "localhost:4001"
 
 
+DEFAULT_GRPC_KEEPALIVE_OPTIONS: tuple[tuple[str, int], ...] = (
+    ("grpc.keepalive_time_ms", 30_000),
+    ("grpc.keepalive_timeout_ms", 10_000),
+    ("grpc.http2.max_pings_without_data", 0),
+    ("grpc.keepalive_permit_without_calls", 1),
+)
+
+
 def get_grpc_channel(
     host_address: Optional[str],
     secure_channel: bool = False,
@@ -81,10 +89,16 @@ def get_grpc_channel(
             host_address = host_address[len(protocol) :]
             break
 
+    merged = dict(DEFAULT_GRPC_KEEPALIVE_OPTIONS)
+    if options:
+        merged.update(dict(options))
+    merged_options = list(merged.items())
     if secure_channel:
-        channel = grpc.secure_channel(host_address, grpc.ssl_channel_credentials(), options=options)
+        channel = grpc.secure_channel(
+            host_address, grpc.ssl_channel_credentials(), options=merged_options
+        )
     else:
-        channel = grpc.insecure_channel(host_address, options=options)
+        channel = grpc.insecure_channel(host_address, options=merged_options)
 
     # Apply interceptors ONLY if they exist
     if interceptors:

durabletask/task.py

@@ -70,13 +70,13 @@ def is_replaying(self) -> bool:
         pass
 
     @abstractmethod
-    def set_custom_status(self, custom_status: Any) -> None:
+    def set_custom_status(self, custom_status: str) -> None:
         """Set the orchestration instance's custom status.
 
         Parameters
         ----------
-        custom_status: Any
-            A JSON-serializable custom status value to set.
+        custom_status: str
+            A custom status string to set.
         """
         pass
 
@@ -396,16 +396,24 @@ class RetryableTask(CompletableTask[T]):
     def __init__(
         self,
         retry_policy: RetryPolicy,
-        action: pb.OrchestratorAction,
         start_time: datetime,
         is_sub_orch: bool,
+        task_name: str,
+        encoded_input: Optional[str] = None,
+        task_execution_id: str = "",
+        instance_id: Optional[str] = None,
+        app_id: Optional[str] = None,
     ) -> None:
         super().__init__()
-        self._action = action
         self._retry_policy = retry_policy
         self._attempt_count = 1
         self._start_time = start_time
         self._is_sub_orch = is_sub_orch
+        self._task_name = task_name
+        self._encoded_input = encoded_input
+        self._task_execution_id = task_execution_id
+        self._instance_id = instance_id
+        self._app_id = app_id
 
     def increment_attempt_count(self) -> None:
         self._attempt_count += 1
@@ -479,9 +487,10 @@ def when_any(tasks: list[Task]) -> WhenAnyTask:
 
 
 class ActivityContext:
-    def __init__(self, orchestration_id: str, task_id: int):
+    def __init__(self, orchestration_id: str, task_id: int, task_execution_id: str = ""):
         self._orchestration_id = orchestration_id
         self._task_id = task_id
+        self._task_execution_id = task_execution_id
 
     @property
     def orchestration_id(self) -> str:
@@ -510,6 +519,21 @@ def task_id(self) -> int:
         """
         return self._task_id
 
+    @property
+    def task_execution_id(self) -> str:
+        """Get the task execution ID associated with this activity invocation.
+
+        The task execution ID is a UUID that is stable across retry attempts
+        of the same activity call. It can be used for idempotency and
+        deduplication when an activity may be retried.
+
+        Returns
+        -------
+        str
+            The task execution ID for this activity invocation.
+        """
+        return self._task_execution_id
+
 
 # Orchestrators are generators that yield tasks and receive/return any type
 Orchestrator = Callable[[OrchestrationContext, TInput], Union[Generator[Task, Any, Any], TOutput]]