fix until WorkflowDefinition

Signed-off-by: Shijie Sheng <liouvetren@gmail.com>

Author: shijiesheng

cadence/_internal/workflow/decision_events_iterator.py

@@ -11,7 +11,6 @@
 
 from cadence._internal.workflow.history_event_iterator import HistoryEventsIterator
 from cadence.api.v1.history_pb2 import HistoryEvent
-from cadence.api.v1.service_worker_pb2 import PollForDecisionTaskResponse
 
 
 @dataclass
@@ -44,10 +43,8 @@ class DecisionEventsIterator(Iterator[DecisionEvents]):
 
     def __init__(
         self,
-        decision_task: PollForDecisionTaskResponse,
         events: List[HistoryEvent],
     ):
-        self._decision_task = decision_task
         self._events: HistoryEventsIterator = HistoryEventsIterator(events)
         self._next_decision_event_id: Optional[int] = None
         self._replay_current_time_milliseconds: Optional[int] = None

cadence/_internal/workflow/workflow_engine.py

@@ -1,5 +1,6 @@
 import logging
 from dataclasses import dataclass
+from typing import List
 
 from cadence._internal.workflow.context import Context
 from cadence._internal.workflow.decision_events_iterator import DecisionEventsIterator
@@ -10,7 +11,10 @@
     CompleteWorkflowExecutionDecisionAttributes,
     Decision,
 )
-from cadence.api.v1.history_pb2 import WorkflowExecutionStartedEventAttributes
+from cadence.api.v1.history_pb2 import (
+    HistoryEvent,
+    WorkflowExecutionStartedEventAttributes,
+)
 from cadence.api.v1.service_worker_pb2 import PollForDecisionTaskResponse
 from cadence.workflow import WorkflowDefinition, WorkflowInfo
 
@@ -33,7 +37,8 @@ def __init__(self, info: WorkflowInfo, workflow_definition: WorkflowDefinition):
         self._context = Context(info, self._decision_manager)
 
     def process_decision(
-        self, decision_task: PollForDecisionTaskResponse
+        self,
+        events: List[HistoryEvent],
     ) -> DecisionResult:
         """
         Process a decision task and generate decisions using DecisionEventsIterator.
@@ -57,18 +62,14 @@ def process_decision(
                         "workflow_type": ctx.info().workflow_type,
                         "workflow_id": ctx.info().workflow_id,
                         "run_id": ctx.info().workflow_run_id,
-                        "started_event_id": decision_task.started_event_id,
-                        "attempt": decision_task.attempt,
                     },
                 )
 
                 # Create DecisionEventsIterator for structured event processing
-                events_iterator = DecisionEventsIterator(
-                    decision_task, ctx.info().workflow_events
-                )
+                events_iterator = DecisionEventsIterator(events)
 
                 # Process decision events using iterator-driven approach
-                self._process_decision_events(ctx, events_iterator, decision_task)
+                self._process_decision_events(ctx, events_iterator)
 
                 # Collect all pending decisions from state machines
                 decisions = self._decision_manager.collect_pending_decisions()
@@ -100,8 +101,6 @@ def process_decision(
                     "workflow_type": ctx.info().workflow_type,
                     "workflow_id": ctx.info().workflow_id,
                     "run_id": ctx.info().workflow_run_id,
-                    "started_event_id": decision_task.started_event_id,
-                    "attempt": decision_task.attempt,
                     "error_type": type(e).__name__,
                 },
                 exc_info=True,
@@ -116,7 +115,6 @@ def _process_decision_events(
         self,
         ctx: Context,
         events_iterator: DecisionEventsIterator,
-        decision_task: PollForDecisionTaskResponse,
     ) -> None:
         """
         Process decision events using the iterator-driven approach similar to Java client.
@@ -139,15 +137,14 @@ def _process_decision_events(
                 "Processing decision events batch",
                 extra={
                     "workflow_id": ctx.info().workflow_id,
-                    "events_count": len(decision_events.get_events()),
-                    "markers_count": len(decision_events.get_markers()),
-                    "replay_mode": decision_events.is_replay(),
+                    "markers_count": len(decision_events.markers),
+                    "replay_mode": decision_events.replay,
                     "replay_time": decision_events.replay_current_time_milliseconds,
                 },
             )
 
             # Update context with replay information
-            ctx.set_replay_mode(decision_events.is_replay())
+            ctx.set_replay_mode(decision_events.replay)
             if decision_events.replay_current_time_milliseconds:
                 ctx.set_replay_current_time_milliseconds(
                     decision_events.replay_current_time_milliseconds
@@ -161,7 +158,7 @@ def _process_decision_events(
                         "workflow_id": ctx.info().workflow_id,
                         "marker_name": getattr(marker_event, "marker_name", "unknown"),
                         "event_id": getattr(marker_event, "event_id", None),
-                        "replay_mode": ctx.is_replay_mode(),
+                        "replay_mode": decision_events.replay,
                     },
                 )
                 # Process through state machines (DecisionsHelper now delegates to DecisionManager)
@@ -175,18 +172,24 @@ def _process_decision_events(
                         "workflow_id": ctx.info().workflow_id,
                         "event_type": getattr(event, "event_type", "unknown"),
                         "event_id": getattr(event, "event_id", None),
-                        "replay_mode": ctx.is_replay_mode(),
+                        "replay_mode": decision_events.replay,
                     },
                 )
+                # start workflow on workflow started event
+                if (
+                    event.WhichOneof("attributes")
+                    == "workflow_execution_started_event_attributes"
+                ):
+                    started_attrs: WorkflowExecutionStartedEventAttributes = (
+                        event.workflow_execution_started_event_attributes
+                    )
+                    if started_attrs and hasattr(started_attrs, "input"):
+                        self._workflow_instance.start(started_attrs.input)
+
                 # Process through state machines (DecisionsHelper now delegates to DecisionManager)
                 self._decision_manager.handle_history_event(event)
 
             # Phase 3: Execute workflow logic
-            if not self._workflow_instance.is_started():
-                self._workflow_instance.start(
-                    self._extract_workflow_input(decision_task)
-                )
-
             self._workflow_instance.run_once()
 
             # Phase 4: update state machine with output events

cadence/_internal/workflow/workflow_intance.py

@@ -12,16 +12,14 @@ def __init__(
     ):
         self._definition = workflow_definition
         self._data_converter = data_converter
-        self._instance = (
-            workflow_definition.cls().__init__()
-        )  # construct a new workflow object
+        self._instance = workflow_definition.cls()  # construct a new workflow object
         self._loop = DeterministicEventLoop()
         self._task: Optional[Task] = None
 
     def start(self, input: Payload):
         if self._task is None:
             run_method = self._definition.get_run_method(self._instance)
-            workflow_input = self._data_converter.from_data(input, [None])
+            workflow_input = self._data_converter.from_data(input, [])
             self._task = self._loop.create_task(run_method(*workflow_input))
 
     def is_started(self) -> bool:

cadence/worker/_decision_task_handler.py

@@ -121,7 +121,6 @@ async def _handle_task_implementation(
             workflow_run_id=run_id,
             workflow_task_list=self.task_list,
             data_converter=self._client.data_converter,
-            workflow_events=workflow_events,
         )
 
         # Use thread-safe cache to get or create workflow engine
@@ -136,7 +135,7 @@ async def _handle_task_implementation(
                 self._workflow_engines[cache_key] = workflow_engine
 
         decision_result = await asyncio.get_running_loop().run_in_executor(
-            self._executor, workflow_engine.process_decision, task
+            self._executor, workflow_engine.process_decision, workflow_events
         )
 
         # Clean up completed workflows from cache to prevent memory leaks

cadence/workflow.py

@@ -5,7 +5,6 @@
 from datetime import timedelta
 from typing import (
     Callable,
-    List,
     cast,
     Optional,
     Union,
@@ -15,10 +14,10 @@
     Type,
     Unpack,
     Any,
+    Generic,
 )
 import inspect
 
-from cadence.api.v1.history_pb2 import HistoryEvent
 from cadence.data_converter import DataConverter
 
 ResultType = TypeVar("ResultType")
@@ -44,6 +43,7 @@ async def execute_activity(
 
 
 T = TypeVar("T", bound=Callable[..., Any])
+C = TypeVar("C")
 
 
 class WorkflowDefinitionOptions(TypedDict, total=False):
@@ -52,16 +52,16 @@ class WorkflowDefinitionOptions(TypedDict, total=False):
     name: str
 
 
-class WorkflowDefinition:
+class WorkflowDefinition(Generic[C]):
     """
     Definition of a workflow class with metadata.
 
     Similar to ActivityDefinition but for workflow classes.
     Provides type safety and metadata for workflow classes.
     """
 
-    def __init__(self, cls: Type, name: str, run_method_name: str):
-        self._cls = cls
+    def __init__(self, cls: Type[C], name: str, run_method_name: str):
+        self._cls: Type[C] = cls
         self._name = name
         self._run_method_name = run_method_name
 
@@ -71,7 +71,7 @@ def name(self) -> str:
         return self._name
 
     @property
-    def cls(self) -> Type:
+    def cls(self) -> Type[C]:
         """Get the workflow class."""
         return self._cls
 
@@ -151,7 +151,7 @@ def decorator(f: T) -> T:
             raise ValueError(f"Workflow run method '{f.__name__}' must be async")
 
         # Attach metadata to the function
-        f._workflow_run = True  # type: ignore
+        setattr(f, "_workflow_run", None)
         return f
 
     # Support both @workflow.run and @workflow.run()
@@ -163,14 +163,13 @@ def decorator(f: T) -> T:
         return decorator(func)
 
 
-@dataclass
+@dataclass(frozen=True)
 class WorkflowInfo:
     workflow_type: str
     workflow_domain: str
     workflow_id: str
     workflow_run_id: str
     workflow_task_list: str
-    workflow_events: List[HistoryEvent]
     data_converter: DataConverter
 
 

tests/cadence/_internal/workflow/test_decision_events_iterator.py

@@ -4,15 +4,12 @@
 """
 
 import pytest
-from typing import List
 
-from cadence.api.v1.history_pb2 import HistoryEvent, History
-from cadence.api.v1.service_worker_pb2 import PollForDecisionTaskResponse
-from cadence.api.v1.common_pb2 import WorkflowExecution
 
 from cadence._internal.workflow.decision_events_iterator import (
     DecisionEventsIterator,
 )
+from tests.cadence._internal.workflow.utils import create_mock_history_event
 
 
 class TestDecisionEventsIterator:
@@ -95,8 +92,7 @@ class TestDecisionEventsIterator:
     )
     def test_successful_cases(self, name, event_types, expected):
         events = create_mock_history_event(event_types)
-        decision_task = create_mock_decision_task(events)
-        iterator = DecisionEventsIterator(decision_task, events)
+        iterator = DecisionEventsIterator(events)
 
         batches = [decision_events for decision_events in iterator]
         assert len(expected) == len(batches)
@@ -108,84 +104,3 @@ def test_successful_cases(self, name, event_types, expected):
             assert batch.replay == expect["replay"]
             assert batch.replay_current_time_milliseconds == expect["replay_time"]
             assert batch.next_decision_event_id == expect["next_decision_event_id"]
-
-
-def create_mock_history_event(event_types: List[str]) -> List[HistoryEvent]:
-    events = []
-    for i, event_type in enumerate(event_types):
-        event = HistoryEvent()
-        event.event_id = i + 1
-        event.event_time.FromMilliseconds((i + 1) * 1000)
-
-        # Set the appropriate attribute based on event type
-        if event_type == "decision_task_started":
-            event.decision_task_started_event_attributes.SetInParent()
-        elif event_type == "decision_task_completed":
-            event.decision_task_completed_event_attributes.SetInParent()
-        elif event_type == "decision_task_failed":
-            event.decision_task_failed_event_attributes.SetInParent()
-        elif event_type == "decision_task_timed_out":
-            event.decision_task_timed_out_event_attributes.SetInParent()
-        elif event_type == "marker_recorded":
-            event.marker_recorded_event_attributes.SetInParent()
-        elif event_type == "activity_scheduled":
-            event.activity_task_scheduled_event_attributes.SetInParent()
-        elif event_type == "activity_started":
-            event.activity_task_started_event_attributes.SetInParent()
-        elif event_type == "activity_completed":
-            event.activity_task_completed_event_attributes.SetInParent()
-        elif event_type == "activity_failed":
-            event.activity_task_failed_event_attributes.SetInParent()
-        elif event_type == "activity_timed_out":
-            event.activity_task_timed_out_event_attributes.SetInParent()
-        elif event_type == "activity_cancel_requested":
-            event.activity_task_cancel_requested_event_attributes.SetInParent()
-        elif event_type == "request_cancel_activity_task_failed":
-            event.request_cancel_activity_task_failed_event_attributes.SetInParent()
-        elif event_type == "activity_canceled":
-            event.activity_task_canceled_event_attributes.SetInParent()
-        elif event_type == "timer_started":
-            event.timer_started_event_attributes.SetInParent()
-        elif event_type == "timer_fired":
-            event.timer_fired_event_attributes.SetInParent()
-        elif event_type == "timer_canceled":
-            event.timer_canceled_event_attributes.SetInParent()
-        elif event_type == "cancel_timer_failed":
-            event.cancel_timer_failed_event_attributes.SetInParent()
-        elif event_type == "request_cancel_external_workflow_execution_initiated":
-            event.request_cancel_external_workflow_execution_initiated_event_attributes.SetInParent()
-        elif event_type == "request_cancel_external_workflow_execution_failed":
-            event.request_cancel_external_workflow_execution_failed_event_attributes.SetInParent()
-        elif event_type == "external_workflow_execution_cancel_requested":
-            event.external_workflow_execution_cancel_requested_event_attributes.SetInParent()
-        elif event_type == "workflow_execution_started":
-            event.workflow_execution_started_event_attributes.SetInParent()
-        elif event_type == "workflow_execution_completed":
-            event.workflow_execution_completed_event_attributes.SetInParent()
-
-        events.append(event)
-
-    return events
-
-
-def create_mock_decision_task(
-    events: List[HistoryEvent], next_page_token: bytes = None
-) -> PollForDecisionTaskResponse:
-    """Create a mock decision task for testing."""
-    task = PollForDecisionTaskResponse()
-
-    # Mock history
-    history = History()
-    history.events.extend(events)
-    task.history.CopyFrom(history)
-
-    # Mock workflow execution
-    workflow_execution = WorkflowExecution()
-    workflow_execution.workflow_id = "test-workflow"
-    workflow_execution.run_id = "test-run"
-    task.workflow_execution.CopyFrom(workflow_execution)
-
-    if next_page_token:
-        task.next_page_token = next_page_token
-
-    return task