fix: address rest of copilot feedback

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

Author: sicoyle

durabletask/task.py

@@ -158,7 +158,7 @@ def call_sub_orchestrator(
         """
         pass
 
-    # TOOD: Add a timeout parameter, which allows the task to be canceled if the event is
+    # TODO: Add a timeout parameter, which allows the task to be canceled if the event is
     # not received within the specified timeout. This requires support for task cancellation.
     @abstractmethod
     def wait_for_external_event(self, name: str) -> Task:
@@ -434,7 +434,22 @@ def compute_next_delay(self) -> Optional[timedelta]:
         else:
             backoff_coefficient = self._retry_policy.backoff_coefficient
 
-        if datetime.utcnow() < retry_expiration:
+        # Compute a deterministic "logical now" based on start time and accumulated delays,
+        # rather than wall-clock time, to avoid non-determinism during replay.
+        total_elapsed_seconds = 0.0
+        for i in range(1, self._attempt_count):
+            attempt_delay = (
+                math.pow(backoff_coefficient, i - 1)
+                * self._retry_policy.first_retry_interval.total_seconds()
+            )
+            if self._retry_policy.max_retry_interval is not None:
+                attempt_delay = min(
+                    attempt_delay,
+                    self._retry_policy.max_retry_interval.total_seconds(),
+                )
+            total_elapsed_seconds += attempt_delay
+        logical_now = self._start_time + timedelta(seconds=total_elapsed_seconds)
+        if logical_now < retry_expiration:
             next_delay_f = (
                 math.pow(backoff_coefficient, self._attempt_count - 1)
                 * self._retry_policy.first_retry_interval.total_seconds()

durabletask_tests/durabletask/test_orchestration_e2e_async.py

@@ -293,8 +293,10 @@ def orchestrator(ctx: task.OrchestrationContext, _):
 
 
 async def test_terminate_recursive():
+    child_instance_id = 'child-instance-id'
+
     def root(ctx: task.OrchestrationContext, _):
-        result = yield ctx.call_sub_orchestrator(child)
+        result = yield ctx.call_sub_orchestrator(child, instance_id=child_instance_id)
         return result
 
     def child(ctx: task.OrchestrationContext, _):
@@ -320,9 +322,9 @@ def child(ctx: task.OrchestrationContext, _):
             assert state.runtime_status == OrchestrationStatus.TERMINATED
 
             # Verify that child orchestration is also terminated
-            await client.wait_for_orchestration_completion(id, timeout=30)
-            assert state is not None
-            assert state.runtime_status == OrchestrationStatus.TERMINATED
+            child_state = await client.wait_for_orchestration_completion(child_instance_id, timeout=30)
+            assert child_state is not None
+            assert child_state.runtime_status == OrchestrationStatus.TERMINATED
 
             await client.purge_orchestration(id)
             state = await client.get_orchestration_state(id)