respond to comment 2

Signed-off-by: Tim Li <[email protected]>

Author: timl3136

cadence/_internal/workflow/workflow_engine.py

@@ -48,8 +48,8 @@ async def process_decision(self, decision_task: PollForDecisionTaskResponse) ->
                     "workflow_type": self._context.info().workflow_type,
                     "workflow_id": self._context.info().workflow_id,
                     "run_id": self._context.info().workflow_run_id,
-                    "started_event_id": getattr(decision_task, 'started_event_id', None),
-                    "attempt": getattr(decision_task, 'attempt', None)
+                    "started_event_id": decision_task.started_event_id,
+                    "attempt": decision_task.attempt
                 }
             )
 
@@ -74,7 +74,7 @@ async def process_decision(self, decision_task: PollForDecisionTaskResponse) ->
                         "workflow_type": self._context.info().workflow_type,
                         "workflow_id": self._context.info().workflow_id,
                         "run_id": self._context.info().workflow_run_id,
-                        "started_event_id": getattr(decision_task, 'started_event_id', None),
+                        "started_event_id": decision_task.started_event_id,
                         "decisions_count": len(decisions),
                         "replay_mode": self._context.is_replay_mode()
                     }
@@ -90,14 +90,14 @@ async def process_decision(self, decision_task: PollForDecisionTaskResponse) ->
                     "workflow_type": self._context.info().workflow_type,
                     "workflow_id": self._context.info().workflow_id,
                     "run_id": self._context.info().workflow_run_id,
-                    "started_event_id": getattr(decision_task, 'started_event_id', None),
-                    "attempt": getattr(decision_task, 'attempt', None),
+                    "started_event_id": decision_task.started_event_id,
+                    "attempt": decision_task.attempt,
                     "error_type": type(e).__name__
                 },
                 exc_info=True
             )
-            # Return empty decisions on error - the task will be failed by the handler
-            return DecisionResult(decisions=[])
+            # Re-raise the exception so the handler can properly handle the failure
+            raise
 
     async def _process_decision_events(self, events_iterator: DecisionEventsIterator, decision_task: PollForDecisionTaskResponse) -> None:
         """

cadence/worker/_decision_task_handler.py

@@ -69,8 +69,8 @@ async def _handle_task_implementation(self, task: PollForDecisionTaskResponse) -
                 "workflow_type": workflow_type_name,
                 "workflow_id": workflow_id,
                 "run_id": run_id,
-                "started_event_id": getattr(task, 'started_event_id', None),
-                "attempt": getattr(task, 'attempt', None),
+                "started_event_id": task.started_event_id,
+                "attempt": task.attempt,
                 "task_token": task.task_token[:16].hex() if task.task_token else None  # Log partial token for debugging
             }
         )
@@ -134,7 +134,7 @@ async def _handle_task_implementation(self, task: PollForDecisionTaskResponse) -
                 "workflow_type": workflow_type_name,
                 "workflow_id": workflow_id,
                 "run_id": run_id,
-                "started_event_id": getattr(task, 'started_event_id', None)
+                "started_event_id": task.started_event_id
             }
         )
 
@@ -159,8 +159,8 @@ async def handle_task_failure(self, task: PollForDecisionTaskResponse, error: Ex
                 "workflow_type": workflow_type,
                 "workflow_id": workflow_id,
                 "run_id": run_id,
-                "started_event_id": getattr(task, 'started_event_id', None),
-                "attempt": getattr(task, 'attempt', None),
+                "started_event_id": task.started_event_id,
+                "attempt": task.attempt,
                 "error_type": type(error).__name__,
                 "error_message": str(error)
             },
@@ -209,7 +209,6 @@ async def handle_task_failure(self, task: PollForDecisionTaskResponse, error: Ex
                 },
                 exc_info=True
             )
-            
 
     async def _respond_decision_task_completed(self, task: PollForDecisionTaskResponse, decision_result: DecisionResult) -> None:
         """
@@ -237,7 +236,7 @@ async def _respond_decision_task_completed(self, task: PollForDecisionTaskRespon
                     "workflow_type": task.workflow_type.name if task.workflow_type else "unknown",
                     "workflow_id": workflow_execution.workflow_id if workflow_execution else "unknown",
                     "run_id": workflow_execution.run_id if workflow_execution else "unknown",
-                    "started_event_id": getattr(task, 'started_event_id', None),
+                    "started_event_id": task.started_event_id,
                     "decisions_count": len(decision_result.decisions),
                     "return_new_decision_task": True,
                     "task_token": task.task_token[:16].hex() if task.task_token else None
@@ -252,7 +251,7 @@ async def _respond_decision_task_completed(self, task: PollForDecisionTaskRespon
                     "workflow_type": task.workflow_type.name if task.workflow_type else "unknown",
                     "workflow_id": workflow_execution.workflow_id if workflow_execution else "unknown",
                     "run_id": workflow_execution.run_id if workflow_execution else "unknown",
-                    "started_event_id": getattr(task, 'started_event_id', None),
+                    "started_event_id": task.started_event_id,
                     "decisions_count": len(decision_result.decisions),
                     "error_type": type(e).__name__
                 },

tests/cadence/worker/test_decision_task_handler.py

@@ -58,6 +58,9 @@ def sample_decision_task(self):
         task.workflow_execution.run_id = "test_run_id"
         task.workflow_type = Mock()
         task.workflow_type.name = "TestWorkflow"
+        # Add the missing attributes that are now accessed directly
+        task.started_event_id = 1
+        task.attempt = 1
         return task
 
     def test_initialization(self, mock_client, mock_registry):
@@ -178,6 +181,8 @@ async def test_handle_task_implementation_different_executions_get_separate_engi
         task1.workflow_execution.run_id = "run_1"
         task1.workflow_type = Mock()
         task1.workflow_type.name = "TestWorkflow"
+        task1.started_event_id = 1
+        task1.attempt = 1
 
         task2 = Mock(spec=PollForDecisionTaskResponse)
         task2.task_token = b"test_task_token_2"
@@ -186,6 +191,8 @@ async def test_handle_task_implementation_different_executions_get_separate_engi
         task2.workflow_execution.run_id = "run_2"          # Different run
         task2.workflow_type = Mock()
         task2.workflow_type.name = "TestWorkflow"
+        task2.started_event_id = 2
+        task2.attempt = 1
 
         # Mock workflow engine
         mock_engine = Mock(spec=WorkflowEngine)

tests/cadence/worker/test_task_handler_integration.py

@@ -53,13 +53,18 @@ def sample_decision_task(self):
         task.workflow_execution.run_id = "test_run_id"
         task.workflow_type = Mock()
         task.workflow_type.name = "TestWorkflow"
+        # Add the missing attributes that are now accessed directly
+        task.started_event_id = 1
+        task.attempt = 1
         return task
 
     @pytest.mark.asyncio
     async def test_full_task_handling_flow_success(self, handler, sample_decision_task, mock_registry):
         """Test the complete task handling flow from base handler through decision handler."""
         # Mock workflow function
-        mock_workflow_func = Mock()
+        def mock_workflow_func(input_data):
+            return f"processed: {input_data}"
+        
         mock_registry.get_workflow.return_value = mock_workflow_func
 
         # Mock workflow engine
@@ -81,7 +86,9 @@ async def test_full_task_handling_flow_success(self, handler, sample_decision_ta
     async def test_full_task_handling_flow_with_error(self, handler, sample_decision_task, mock_registry):
         """Test the complete task handling flow when an error occurs."""
         # Mock workflow function
-        mock_workflow_func = Mock()
+        def mock_workflow_func(input_data):
+            return f"processed: {input_data}"
+        
         mock_registry.get_workflow.return_value = mock_workflow_func
 
         # Mock workflow engine to raise an error
@@ -102,7 +109,9 @@ async def test_full_task_handling_flow_with_error(self, handler, sample_decision
     async def test_context_activation_integration(self, handler, sample_decision_task, mock_registry):
         """Test that context activation works correctly in the integration."""
         # Mock workflow function
-        mock_workflow_func = Mock()
+        def mock_workflow_func(input_data):
+            return f"processed: {input_data}"
+        
         mock_registry.get_workflow.return_value = mock_workflow_func
 
         # Mock workflow engine
@@ -133,7 +142,9 @@ def track_context_activation():
     async def test_multiple_workflow_executions(self, handler, mock_registry):
         """Test handling multiple workflow executions creates new engines for each."""
         # Mock workflow function
-        mock_workflow_func = Mock()
+        def mock_workflow_func(input_data):
+            return f"processed: {input_data}"
+        
         mock_registry.get_workflow.return_value = mock_workflow_func
 
         # Create multiple decision tasks for different workflows
@@ -144,6 +155,8 @@ async def test_multiple_workflow_executions(self, handler, mock_registry):
         task1.workflow_execution.run_id = "run1"
         task1.workflow_type = Mock()
         task1.workflow_type.name = "TestWorkflow"
+        task1.started_event_id = 1
+        task1.attempt = 1
 
         task2 = Mock(spec=PollForDecisionTaskResponse)
         task2.task_token = b"task2_token"
@@ -152,6 +165,8 @@ async def test_multiple_workflow_executions(self, handler, mock_registry):
         task2.workflow_execution.run_id = "run2"
         task2.workflow_type = Mock()
         task2.workflow_type.name = "TestWorkflow"
+        task2.started_event_id = 2
+        task2.attempt = 1
 
         # Mock workflow engine
         mock_engine = Mock(spec=WorkflowEngine)
@@ -176,7 +191,9 @@ async def test_multiple_workflow_executions(self, handler, mock_registry):
     async def test_workflow_engine_creation_integration(self, handler, sample_decision_task, mock_registry):
         """Test workflow engine creation integration."""
         # Mock workflow function
-        mock_workflow_func = Mock()
+        def mock_workflow_func(input_data):
+            return f"processed: {input_data}"
+        
         mock_registry.get_workflow.return_value = mock_workflow_func
 
         # Mock workflow engine
@@ -197,7 +214,9 @@ async def test_workflow_engine_creation_integration(self, handler, sample_decisi
     async def test_error_handling_with_context_cleanup(self, handler, sample_decision_task, mock_registry):
         """Test that context cleanup happens even when errors occur."""
         # Mock workflow function
-        mock_workflow_func = Mock()
+        def mock_workflow_func(input_data):
+            return f"processed: {input_data}"
+        
         mock_registry.get_workflow.return_value = mock_workflow_func
 
         # Mock workflow engine to raise an error
@@ -231,7 +250,9 @@ async def test_concurrent_task_handling(self, handler, mock_registry):
         import asyncio
 
         # Mock workflow function
-        mock_workflow_func = Mock()
+        def mock_workflow_func(input_data):
+            return f"processed: {input_data}"
+        
         mock_registry.get_workflow.return_value = mock_workflow_func
 
         # Create multiple tasks
@@ -244,6 +265,8 @@ async def test_concurrent_task_handling(self, handler, mock_registry):
             task.workflow_execution.run_id = f"run{i}"
             task.workflow_type = Mock()
             task.workflow_type.name = "TestWorkflow"
+            task.started_event_id = i + 1
+            task.attempt = 1
             tasks.append(task)
 
         # Mock workflow engine