cleanup/feedback

Signed-off-by: Filinto Duran <[email protected]>

Author: filintod

durabletask/aio/__init__.py

@@ -27,7 +27,7 @@
 from .compatibility import OrchestrationContextProtocol, ensure_compatibility
 
 # Core context and driver
-from .context import AsyncWorkflowContext, WorkflowInfo
+from .context import AsyncWorkflowContext
 from .driver import CoroutineOrchestratorRunner, WorkflowFunction
 
 # Sandbox and error handling
@@ -38,20 +38,12 @@
     WorkflowTimeoutError,
     WorkflowValidationError,
 )
-from .sandbox import (
-    SandboxMode,
-    _NonDeterminismDetector,
-    sandbox_best_effort,
-    sandbox_off,
-    sandbox_scope,
-    sandbox_strict,
-)
+from .sandbox import SandboxMode, _NonDeterminismDetector
 
 __all__ = [
     "AsyncTaskHubGrpcClient",
     # Core classes
     "AsyncWorkflowContext",
-    "WorkflowInfo",
     "CoroutineOrchestratorRunner",
     "WorkflowFunction",
     # Deterministic utilities
@@ -73,11 +65,7 @@
     "SwallowExceptionAwaitable",
     "gather",
     # Sandbox and utilities
-    "sandbox_scope",
     "SandboxMode",
-    "sandbox_off",
-    "sandbox_best_effort",
-    "sandbox_strict",
     "_NonDeterminismDetector",
     # Compatibility protocol
     "OrchestrationContextProtocol",

durabletask/aio/awaitables.py

@@ -378,7 +378,7 @@ def __await__(self) -> Generator[Any, Any, List[TOutput]]:
 class WhenAnyAwaitable(AwaitableBase[task.Task[Any]]):
     """Awaitable for when_any operations (wait for any task to complete)."""
 
-    __slots__ = ("_tasks_like",)
+    __slots__ = ("_originals", "_underlying")
 
     def __init__(self, tasks_like: Iterable[Union[AwaitableBase[Any], task.Task[Any]]]):
         """
@@ -388,33 +388,33 @@ def __init__(self, tasks_like: Iterable[Union[AwaitableBase[Any], task.Task[Any]
             tasks_like: Iterable of awaitables or tasks to wait for
         """
         super().__init__()
-        self._tasks_like = list(tasks_like)
+        self._originals = list(tasks_like)
+        # Defer conversion to avoid issues with incomplete mocks and coroutine reuse
+        self._underlying: Optional[List[task.Task[Any]]] = None
+
+    def _ensure_underlying(self) -> List[task.Task[Any]]:
+        """Lazily convert originals to tasks, caching the result."""
+        if self._underlying is None:
+            self._underlying = []
+            for a in self._originals:
+                if isinstance(a, AwaitableBase):
+                    self._underlying.append(a._to_task())
+                elif isinstance(a, task.Task):
+                    self._underlying.append(a)
+                else:
+                    raise TypeError("when_any expects AwaitableBase or durabletask.task.Task")
+        return self._underlying
 
     def _to_task(self) -> task.Task[Any]:
         """Convert to a when_any task."""
-        underlying: List[task.Task[Any]] = []
-        for a in self._tasks_like:
-            if isinstance(a, AwaitableBase):
-                underlying.append(a._to_task())
-            elif isinstance(a, task.Task):
-                underlying.append(a)
-            else:
-                raise TypeError("when_any expects AwaitableBase or durabletask.task.Task")
-        return cast(task.Task[Any], task.when_any(underlying))
+        return cast(task.Task[Any], task.when_any(self._ensure_underlying()))
 
     def __await__(self) -> Generator[Any, Any, Any]:
         """Return a proxy that compares equal to the original item and exposes get_result()."""
-        when_any_task = self._to_task()
+        underlying = self._ensure_underlying()
+        when_any_task = task.when_any(underlying)
         completed = yield when_any_task
 
-        # Build underlying mapping original -> underlying task
-        underlying: List[task.Task[Any]] = []
-        for a in self._tasks_like:
-            if isinstance(a, AwaitableBase):
-                underlying.append(a._to_task())
-            elif isinstance(a, task.Task):
-                underlying.append(a)
-
         class _CompletedProxy:
             __slots__ = ("_original", "_completed")
 
@@ -431,20 +431,28 @@ def get_result(self) -> Any:
                     return self._completed.get_result()
                 return getattr(self._completed, "result", None)
 
+            @property
+            def __dict__(self) -> dict[str, Any]:
+                """Expose a dict-like view for compatibility with user code."""
+                return {
+                    "_original": self._original,
+                    "_completed": self._completed,
+                }
+
             def __repr__(self) -> str:  # pragma: no cover
                 return f"<WhenAnyCompleted proxy for {self._original!r}>"
 
         # If the runtime returned a non-task sentinel (e.g., tests), assume first item won
         if not isinstance(completed, task.Task):
-            return _CompletedProxy(self._tasks_like[0], completed)
+            return _CompletedProxy(self._originals[0], completed)
 
         # Map completed task back to the original item and return proxy
-        for original, under in zip(self._tasks_like, underlying, strict=False):
+        for original, under in zip(self._originals, underlying, strict=False):
             if completed == under:
                 return _CompletedProxy(original, completed)
 
         # Fallback proxy; treat the first as original
-        return _CompletedProxy(self._tasks_like[0], completed)
+        return _CompletedProxy(self._originals[0], completed)
 
 
 class WhenAnyResultAwaitable(AwaitableBase[tuple[int, Any]]):
@@ -454,7 +462,7 @@ class WhenAnyResultAwaitable(AwaitableBase[tuple[int, Any]]):
     This is useful when you need to know which task completed first, not just its result.
     """
 
-    __slots__ = ("_tasks_like", "_awaitables")
+    __slots__ = ("_originals", "_underlying")
 
     def __init__(self, tasks_like: Iterable[Union[AwaitableBase[Any], task.Task[Any]]]):
         """
@@ -464,41 +472,37 @@ def __init__(self, tasks_like: Iterable[Union[AwaitableBase[Any], task.Task[Any]
             tasks_like: Iterable of awaitables or tasks to wait for
         """
         super().__init__()
-        self._tasks_like = list(tasks_like)
-        self._awaitables = self._tasks_like  # Alias for compatibility
+        self._originals = list(tasks_like)
+        # Defer conversion to avoid issues with incomplete mocks and coroutine reuse
+        self._underlying: Optional[List[task.Task[Any]]] = None
+
+    def _ensure_underlying(self) -> List[task.Task[Any]]:
+        """Lazily convert originals to tasks, caching the result."""
+        if self._underlying is None:
+            self._underlying = []
+            for a in self._originals:
+                if isinstance(a, AwaitableBase):
+                    self._underlying.append(a._to_task())
+                elif isinstance(a, task.Task):
+                    self._underlying.append(a)
+                else:
+                    raise TypeError(
+                        "when_any_with_result expects AwaitableBase or durabletask.task.Task"
+                    )
+        return self._underlying
 
     def _to_task(self) -> task.Task[Any]:
         """Convert to a when_any task with result tracking."""
-        underlying: List[task.Task[Any]] = []
-        for a in self._tasks_like:
-            if isinstance(a, AwaitableBase):
-                underlying.append(a._to_task())
-            elif isinstance(a, task.Task):
-                underlying.append(a)
-            else:
-                raise TypeError(
-                    "when_any_with_result expects AwaitableBase or durabletask.task.Task"
-                )
-
-        # Use when_any and then determine which task completed
-        when_any_task = task.when_any(underlying)
-        return cast(task.Task[Any], when_any_task)
+        return cast(task.Task[Any], task.when_any(self._ensure_underlying()))
 
     def __await__(self) -> Generator[Any, Any, tuple[int, Any]]:
         """Override to provide index + result tuple."""
-        t = self._to_task()
-        completed_task = yield t
-
-        # Find which task completed by comparing results
-        underlying_tasks: List[task.Task[Any]] = []
-        for a in self._tasks_like:
-            if isinstance(a, AwaitableBase):
-                underlying_tasks.append(a._to_task())
-            elif isinstance(a, task.Task):
-                underlying_tasks.append(a)
+        underlying = self._ensure_underlying()
+        when_any_task = task.when_any(underlying)
+        completed_task = yield when_any_task
 
         # The completed_task should match one of our underlying tasks
-        for i, underlying_task in enumerate(underlying_tasks):
+        for i, underlying_task in enumerate(underlying):
             if underlying_task == completed_task:
                 return (i, completed_task.result if hasattr(completed_task, "result") else None)
 

durabletask/aio/compatibility.py

@@ -54,16 +54,6 @@ def workflow_name(self) -> Optional[str]:
         """Get the orchestrator name/type for this instance."""
         ...
 
-    @property
-    def parent_instance_id(self) -> Optional[str]:
-        """Get the parent orchestration ID if this is a sub-orchestration."""
-        ...
-
-    @property
-    def history_event_sequence(self) -> Optional[int]:
-        """Get the current processed history event sequence."""
-        ...
-
     @property
     def is_suspended(self) -> bool:
         """Get whether this orchestration is currently suspended."""
@@ -132,8 +122,6 @@ def ensure_compatibility(context_class: type) -> type:
         "current_utc_datetime",
         "is_replaying",
         "workflow_name",
-        "parent_instance_id",
-        "history_event_sequence",
         "is_suspended",
     ]
 

durabletask/aio/context.py

@@ -20,7 +20,6 @@
 from __future__ import annotations
 
 import os
-from dataclasses import dataclass
 from datetime import datetime, timedelta
 from typing import Any, Callable, Dict, List, Optional, TypeVar, Union, cast
 
@@ -45,24 +44,6 @@
 T = TypeVar("T")
 
 
-@dataclass(frozen=True)
-class WorkflowInfo:
-    """
-    Read-only metadata snapshot about the running workflow execution.
-
-    Similar to Temporal's workflow.info, this provides convenient access to
-    workflow execution metadata in a single immutable object.
-    """
-
-    instance_id: str
-    workflow_name: Optional[str]
-    is_replaying: bool
-    is_suspended: bool
-    parent_instance_id: Optional[str]
-    current_time: datetime
-    history_event_sequence: int
-
-
 @ensure_compatibility
 class AsyncWorkflowContext(DeterministicContextMixin):
     """
@@ -130,52 +111,15 @@ def is_replaying(self) -> bool:
     @property
     def is_suspended(self) -> bool:
         """Check if the workflow is currently suspended."""
-        return getattr(self._base_ctx, "is_suspended", False)
+        return self._base_ctx.is_suspended
 
     @property
     def workflow_name(self) -> Optional[str]:
         """Get the workflow name."""
         return getattr(self._base_ctx, "workflow_name", None)
 
-    @property
-    def parent_instance_id(self) -> Optional[str]:
-        """Get the parent instance ID (for sub-orchestrators)."""
-        return getattr(self._base_ctx, "parent_instance_id", None)
-
-    @property
-    def history_event_sequence(self) -> int:
-        """Get the current history event sequence number."""
-        return getattr(self._base_ctx, "history_event_sequence", 0)
-
-    @property
-    def execution_info(self) -> Optional[Any]:
-        """Get execution_info from the base context if available, else None."""
-        return getattr(self._base_ctx, "execution_info", None)
-
-    @property
-    def info(self) -> WorkflowInfo:
-        """
-        Get a read-only snapshot of workflow execution metadata.
-
-        This provides a Temporal-style info object bundling instance_id, workflow_name,
-        is_replaying, timestamps, and other metadata in a single immutable object.
-        Useful for deterministic logging, idempotency keys, and conditional logic based on replay state.
-
-        Returns:
-            WorkflowInfo: Immutable dataclass with workflow execution metadata
-        """
-        return WorkflowInfo(
-            instance_id=self.instance_id,
-            workflow_name=self.workflow_name,
-            is_replaying=self.is_replaying,
-            is_suspended=self.is_suspended,
-            parent_instance_id=self.parent_instance_id,
-            current_time=self.current_utc_datetime,
-            history_event_sequence=self.history_event_sequence,
-        )
-
     # Activity operations
-    def activity(
+    def call_activity(
         self,
         activity_fn: Union[dt_task.Activity[Any, Any], str],
         *,
@@ -206,24 +150,6 @@ def activity(
             metadata=metadata,
         )
 
-    def call_activity(
-        self,
-        activity_fn: Union[dt_task.Activity[Any, Any], str],
-        *,
-        input: Any = None,
-        retry_policy: Any = None,
-        app_id: Optional[str] = None,
-        metadata: Optional[Dict[str, str]] = None,
-    ) -> ActivityAwaitable[Any]:
-        """Alias for activity() method for API compatibility."""
-        return self.activity(
-            activity_fn,
-            input=input,
-            retry_policy=retry_policy,
-            app_id=app_id,
-            metadata=metadata,
-        )
-
     # Sub-orchestrator operations
     def sub_orchestrator(
         self,