Skip to content

feat!: Return Futures Instead of Coroutines #199

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
natemort wants to merge 1 commit into
base: main
Choose a base branch
from
+883 −728
Draft

feat!: Return Futures Instead of Coroutines #199

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 8 additions & 3 deletions cadence/_internal/activity/_definition.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,12 @@
)

from cadence._internal.fn_signature import FnSignature
from cadence.workflow import ActivityOptions, WorkflowContext, execute_activity
from cadence.workflow import (
ActivityOptions,
WorkflowContext,
execute_activity,
WorkflowFuture,
)

T = TypeVar("T")
P = ParamSpec("P")
Expand Down Expand Up @@ -80,9 +85,9 @@ def with_options(self, **kwargs: Unpack[ActivityOptions]) -> Self:
res._execution_options = new_opts
return res

async def execute(self, *args: P.args, **kwargs: P.kwargs) -> R:
def execute(self, *args: P.args, **kwargs: P.kwargs) -> WorkflowFuture[R]:

@gitar-bot gitar-bot Bot Oct 15, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Cross-Repo: ai-skills TaskGroup fan-out example breaks: execute() returns a Future

Fix here and in cadence-workflow/ai-skills

Breaks in: cadence-workflow/ai-skills - skills/cadence-developer/knowledge/python/workflows.md - parallel fan-out example with asyncio.TaskGroup and tg.create_task(...execute(...))

This PR changes ActivityDefinition.execute to return a WorkflowFuture, which is an asyncio.Future subclass, where it used to return a coroutine. The change is intentional (feat!). The guidance in cadence-workflow/ai-skills, in skills/cadence-developer/knowledge/python/workflows.md, still shows parallel fan-out as tg.create_task(parallel_one.with_options(...).execute("1")) inside asyncio.TaskGroup. create_task only accepts coroutines and raises TypeError when given a Future, so workflows written from that skill will crash on first execution. This PR's own nondeterminism integration tests were migrated from TaskGroup/create_task to asyncio.gather(...) for the same reason. The consumer has to move. On this side, the breaking change should be called out in the release notes so dependent docs and skills get updated.

Here: Keep the new return type. Add a release note or changelog entry (the PR template's 'Release notes' section is empty) saying that execute(), execute_child_workflow(), sleep(), wait_condition() and the signal APIs now return WorkflowFuture. Code must use await, asyncio.gather(...), or .cancel() on the future directly, because asyncio.create_task(...)/TaskGroup.create_task(...) no longer accept these values.

There: In skills/cadence-developer/knowledge/python/workflows.md, rewrite the parallel fan-out example to await the returned futures with asyncio.gather. Do not wrap them in tg.create_task(...).

In cadence-workflow/ai-skills - skills/cadence-developer/knowledge/python/workflows.md:

results = await asyncio.gather(
    parallel_one.with_options(
        schedule_to_close_timeout=timedelta(seconds=10)
    ).execute("1"),
    parallel_two.with_options(
        schedule_to_close_timeout=timedelta(seconds=10)
    ).execute("2"),
)

Was this helpful? React with 👍 / 👎

result_type = cast(type[R], self._signature.return_type)
return await execute_activity(
return execute_activity(
self._name,
result_type,
*self._signature.params_from_call(args, kwargs),
Expand Down
62 changes: 32 additions & 30 deletions cadence/_internal/workflow/child_workflow_future.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,44 +1,46 @@
import asyncio
from typing import Any, Awaitable, Generator, Type, cast

from cadence.api.v1.common_pb2 import Payload
from cadence.data_converter import DataConverter
from cadence.workflow import ResultType, WorkflowContext


class ChildWorkflowFutureImpl(Awaitable[ResultType]):
from typing import Any

from cadence._internal.workflow.statemachine.decision_state_machine import (
CancelFn,
ConverterFn,
DecisionFuture,
)
from cadence.api.v1.common_pb2 import WorkflowExecution
from cadence.workflow import (
ChildWorkflowFuture,
ResultType,
WorkflowContext,
WorkflowFuture,
)


class ChildWorkflowFutureImpl(
DecisionFuture[ResultType], ChildWorkflowFuture[ResultType]
):
def __init__(
self,
loop: asyncio.AbstractEventLoop | None = None,
request_cancel: CancelFn | None = None,
converter: ConverterFn[ResultType] | None = None,
*,
workflow_id: str,
run_id: str,
result_future: "asyncio.Future[Payload]",
result_type: Type[ResultType],
data_converter: DataConverter,
execution: WorkflowFuture[WorkflowExecution],
) -> None:
super().__init__(loop, request_cancel, converter)
self._workflow_id = workflow_id
self._run_id = run_id
self._result_future = result_future
self._result_type = result_type
self._data_converter = data_converter
self._execution = execution

@property
def workflow_id(self) -> str:
return self._workflow_id

@property
def run_id(self) -> str:
return self._run_id
def execution(self) -> WorkflowFuture[WorkflowExecution]:
return self._execution

def cancel(self) -> bool:
"""Request cancellation of the child workflow."""
return self._result_future.cancel()

async def signal(self, signal_name: str, *args: Any) -> None:
def signal(self, signal_name: str, *args: Any) -> WorkflowFuture[None]:
"""Send a signal to this child workflow."""
ctx = WorkflowContext.get()
await ctx.signal_child_workflow(self._workflow_id, signal_name, *args)

def __await__(self) -> Generator[Any, None, ResultType]:
payload: Payload = yield from self._result_future.__await__()
result = self._data_converter.from_data(payload, [self._result_type])[0]
return cast(ResultType, result)
return WorkflowContext.get().signal_child_workflow(
self._workflow_id, signal_name, *args
)
76 changes: 27 additions & 49 deletions cadence/_internal/workflow/context.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,6 @@
from math import ceil
from typing import Iterator, Optional, Any, Unpack, Type, cast, Callable, Mapping

from cadence._internal.workflow.child_workflow_future import ChildWorkflowFutureImpl
from cadence._internal.workflow.deterministic_event_loop import DeterministicEventLoop
from cadence._internal.workflow.deterministic_event_loop import FatalDecisionError
from cadence._internal.workflow.memo import memo_to_proto
Expand Down Expand Up @@ -51,6 +50,7 @@
SearchAttributeType,
WorkflowCancellationInfo,
WorkflowContext,
WorkflowFuture,
WorkflowInfo,
DEFAULT_VERSION,
)
Expand Down Expand Up @@ -83,13 +83,13 @@ def info(self) -> WorkflowInfo:
def data_converter(self) -> DataConverter:
return self.info().data_converter

async def execute_activity(
def execute_activity(
self,
activity: str,
result_type: Type[ResultType],
*args: Any,
**kwargs: Unpack[ActivityOptions],
) -> ResultType:
) -> WorkflowFuture[ResultType]:
opts: ActivityOptions = {**_DEFAULT_ACTIVITY_OPTIONS, **kwargs}
if "schedule_to_close_timeout" not in opts and (
"schedule_to_start_timeout" not in opts
Expand Down Expand Up @@ -137,26 +137,11 @@ async def execute_activity(
)
set_header(schedule_attributes, self._context_propagators)

future = self._decision_manager.schedule_activity(schedule_attributes)
result_payload = await future

result = self.data_converter().from_data(result_payload, [result_type])[0]

return cast(ResultType, result)

async def execute_child_workflow(
self,
workflow_type: str,
result_type: Type[ResultType],
*args: Any,
**kwargs: Unpack[ChildWorkflowOptions],
) -> ResultType:
future = await self.start_child_workflow(
workflow_type, result_type, *args, **kwargs
return self._decision_manager.schedule_activity(
schedule_attributes, self.data_converter(), result_type
)
return await future

async def start_child_workflow(
def execute_child_workflow(
self,
workflow_type: str,
result_type: Type[ResultType],
Expand All @@ -166,19 +151,11 @@ async def start_child_workflow(
schedule_attributes = self._build_child_workflow_attrs(
workflow_type, *args, **kwargs
)
execution_future, result_future = (
self._decision_manager.schedule_child_workflow(
schedule_attributes,
parent_workflow_run_id=self._info.workflow_run_id,
)
)
workflow_execution = await execution_future
return ChildWorkflowFutureImpl(
workflow_id=workflow_execution.workflow_id,
run_id=workflow_execution.run_id,
result_future=result_future,
result_type=result_type,
data_converter=self.data_converter(),
return self._decision_manager.schedule_child_workflow(
schedule_attributes,
self.data_converter(),
result_type,
parent_workflow_run_id=self._info.workflow_run_id,
)

def _build_child_workflow_attrs(
Expand Down Expand Up @@ -244,39 +221,40 @@ def _build_child_workflow_attrs(

return schedule_attributes

async def signal_child_workflow(
def signal_child_workflow(
self,
child_workflow_id: str,
signal_name: str,
*args: Any,
) -> None:
) -> WorkflowFuture[None]:
if not child_workflow_id:
raise ValueError("child_workflow_id must not be empty")
await self._signal_workflow(
return self._signal_workflow(
child_workflow_id, signal_name, args, child_workflow_only=True
)

async def signal_external_workflow(
def signal_external_workflow(
self,
workflow_id: str,
signal_name: str,
*args: Any,
run_id: str = "",
domain: str = "",
) -> None:
) -> WorkflowFuture[None]:
if not workflow_id:
raise ValueError("workflow_id must not be empty")
await self._signal_workflow(
return self._signal_workflow(
workflow_id, signal_name, args, run_id=run_id, domain=domain
)

async def start_timer(self, duration: timedelta):
def start_timer(self, duration: timedelta) -> WorkflowFuture[None]:
if duration.total_seconds() <= 0: # shortcut
return
res = WorkflowFuture[None]()
res.set_result(None)
return res
attributes = StartTimerDecisionAttributes()
attributes.start_to_fire_timeout.FromTimedelta(duration)
future = self._decision_manager.start_timer(attributes)
await future
return self._decision_manager.start_timer(attributes)

def set_replay_mode(self, replay: bool) -> None:
"""Set whether the workflow is currently in replay mode."""
Expand Down Expand Up @@ -413,9 +391,9 @@ def get_replay_current_time(self) -> Optional[datetime]:
"""Get the current replay timestamp."""
return self._replay_current_time

async def wait_condition(self, predicate: Callable[[], bool]) -> None:
def wait_condition(self, predicate: Callable[[], bool]) -> WorkflowFuture[None]:
loop = cast(DeterministicEventLoop, get_running_loop())
await loop.create_waiter(predicate)
return loop.create_waiter(predicate)

def request_cancel(
self, attrs: WorkflowExecutionCancelRequestedEventAttributes
Expand All @@ -433,7 +411,7 @@ def is_cancel_requested(self) -> bool:
def inject_propagated_headers(self) -> dict[str, bytes]:
return inject_headers(self._context_propagators)

async def _signal_workflow(
def _signal_workflow(
self,
workflow_id: str,
signal_name: str,
Expand All @@ -442,7 +420,7 @@ async def _signal_workflow(
run_id: str = "",
domain: str = "",
child_workflow_only: bool = False,
) -> None:
) -> WorkflowFuture[None]:
if not signal_name:
raise ValueError("signal_name must not be empty")
attrs = SignalExternalWorkflowExecutionDecisionAttributes(
Expand All @@ -455,7 +433,7 @@ async def _signal_workflow(
input=self.data_converter().to_data(list(args)),
child_workflow_only=child_workflow_only,
)
await self._decision_manager.signal_external_workflow(attrs)
return self._decision_manager.signal_external_workflow(attrs)


def _round_to_nearest_second(delta: timedelta) -> timedelta:
Expand Down
9 changes: 5 additions & 4 deletions cadence/_internal/workflow/statemachine/activity_state_machine.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
from typing import Any

from cadence._internal.workflow.statemachine.decision_state_machine import (
DecisionState,
DecisionType,
Expand All @@ -10,7 +12,6 @@
record_immediate_cancel,
)
from cadence.api.v1 import decision, history
from cadence.api.v1.common_pb2 import Payload
from cadence.api.v1.workflow_pb2 import TimeoutType
from cadence.error import ActivityFailure

Expand All @@ -19,12 +20,12 @@

class ActivityStateMachine(BaseDecisionStateMachine):
request: decision.ScheduleActivityTaskDecisionAttributes
completed: DecisionFuture[Payload]
completed: DecisionFuture[Any]

def __init__(
self,
request: decision.ScheduleActivityTaskDecisionAttributes,
completed: DecisionFuture[Payload],
completed: DecisionFuture[Any],
) -> None:
super().__init__()
self.request = request
Expand Down Expand Up @@ -82,7 +83,7 @@ def handle_completed(
self, event: history.ActivityTaskCompletedEventAttributes
) -> None:
self._transition(DecisionState.COMPLETED)
self._resolve(self.completed, result=event.result)
self.completed.complete_with_payload(event.result)

@gitar-bot gitar-bot Bot Oct 15, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ [cross-pr] Merging with #196 loses typed activity results or ActivityError wrapping

Breaks when #199 and #196 both merge (in either order) and the state machine and DecisionManager hunks are combined mechanically. Any activity or child workflow that completes or fails then hits the mismatch.

How

  1. feat!: Return Futures Instead of Coroutines #199 makes ActivityStateMachine.handle_completed call self.completed.complete_with_payload(event.result) (cadence/_internal/workflow/statemachine/activity_state_machine.py:86). DecisionManager.schedule_activity(attrs, converter, result_type) passes a ConverterFn into DecisionFuture (decision_manager.py:159-172).
  2. feat!: Standardize Error Handling and Support Custom Error Reasons #196 gives ActivityStateMachine a third data_converter constructor argument. Its handle_failed/handle_timeout build errors with activity_failed_error(..., self._data_converter), and it resolves handle_completed with the raw Payload via _resolve.
  3. feat!: Standardize Error Handling and Support Custom Error Reasons #196's schedule_activity builds ActivityStateMachine(attrs, future, self._data_converter) with a plain future. feat!: Return Futures Instead of Coroutines #199 builds ActivityStateMachine(attrs, future) with a converting future.
  4. Combining the two means either handle_failed raises AttributeError on _data_converter, or await execute_activity(..., str) yields a raw Payload. The same split affects ChildWorkflowExecutionStateMachine and schedule_child_workflow.

Fix

Compared e7f8d08 with #196 at 59c684d

Was this helpful? React with 👍 / 👎


@activity_events.event()
def handle_failed(self, event: history.ActivityTaskFailedEventAttributes) -> None:
Expand Down
10 changes: 6 additions & 4 deletions cadence/_internal/workflow/statemachine/child_workflow_execution_state_machine.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
from __future__ import annotations

from typing import Any

from cadence._internal.workflow.statemachine.decision_state_machine import (
BaseDecisionStateMachine,
DecisionFuture,
Expand All @@ -12,7 +14,7 @@
record_immediate_cancel,
)
from cadence.api.v1 import decision, history
from cadence.api.v1.common_pb2 import Payload, WorkflowExecution
from cadence.api.v1.common_pb2 import WorkflowExecution
from cadence.error import (
ChildWorkflowExecutionCanceled,
ChildWorkflowExecutionFailed,
Expand All @@ -29,14 +31,14 @@ class ChildWorkflowExecutionStateMachine(BaseDecisionStateMachine):

request: decision.StartChildWorkflowExecutionDecisionAttributes
execution: DecisionFuture[WorkflowExecution]
result: DecisionFuture[Payload]
result: DecisionFuture[Any]
_run_id: str | None

def __init__(
self,
request: decision.StartChildWorkflowExecutionDecisionAttributes,
execution: DecisionFuture[WorkflowExecution],
result: DecisionFuture[Payload],
result: DecisionFuture[Any],
) -> None:
super().__init__()
self.request = request
Expand Down Expand Up @@ -130,7 +132,7 @@ def handle_completed(
self, event: history.ChildWorkflowExecutionCompletedEventAttributes
) -> None:
self._transition(DecisionState.COMPLETED)
self._resolve(self.result, result=event.result)
self.result.complete_with_payload(event.result)

@child_workflow_events.event()
def handle_failed(
Expand Down
Loading
Loading