respond to comments

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

Author: timl3136

cadence/_internal/workflow/workflow_engine.py

@@ -11,8 +11,6 @@
 @dataclass
 class DecisionResult:
     decisions: list[Decision]
-    force_create_new_decision_task: bool = False
-    query_results: Optional[dict] = None
 
 class WorkflowEngine:
     def __init__(self, info: WorkflowInfo, client: Client, workflow_func: Optional[Callable[..., Any]] = None):

cadence/worker/_decision_task_handler.py

@@ -36,7 +36,7 @@ def __init__(self, client: Client, task_list: str, registry: Registry, identity:
         """
         super().__init__(client, task_list, identity, **options)
         self._registry = registry
-        self._workflow_engines: Dict[str, WorkflowEngine] = {}
+        self._workflow_engine: WorkflowEngine
 
 
     async def _handle_task_implementation(self, task: PollForDecisionTaskResponse) -> None:
@@ -51,7 +51,7 @@ async def _handle_task_implementation(self, task: PollForDecisionTaskResponse) -
         workflow_type = task.workflow_type
 
         if not workflow_execution or not workflow_type:
-            logger.error("Decision task missing workflow execution or type")
+            logger.error("Decision task missing workflow execution or type. Task: %r", task)
             raise ValueError("Missing workflow execution or type")
 
         workflow_id = workflow_execution.workflow_id
@@ -60,34 +60,27 @@ async def _handle_task_implementation(self, task: PollForDecisionTaskResponse) -
 
         logger.info(f"Processing decision task for workflow {workflow_id} (type: {workflow_type_name})")
 
-        # Get or create workflow engine for this workflow execution
-        engine_key = f"{workflow_id}:{run_id}"
-        if engine_key not in self._workflow_engines:
-            # Get the workflow function from registry
-            try:
-                workflow_func = self._registry.get_workflow(workflow_type_name)
-            except KeyError:
-                logger.error(f"Workflow type '{workflow_type_name}' not found in registry")
-                raise KeyError(f"Workflow type '{workflow_type_name}' not found")
-            
-            # Create workflow info and engine
-            workflow_info = WorkflowInfo(
-                workflow_type=workflow_type_name,
-                workflow_domain=self._client.domain,
-                workflow_id=workflow_id,
-                workflow_run_id=run_id
-            )
-            
-            self._workflow_engines[engine_key] = WorkflowEngine(
-                info=workflow_info, 
-                client=self._client, 
-                workflow_func=workflow_func
-            )
+        try:
+            workflow_func = self._registry.get_workflow(workflow_type_name)
+        except KeyError:
+            logger.error(f"Workflow type '{workflow_type_name}' not found in registry")
+            raise KeyError(f"Workflow type '{workflow_type_name}' not found")
+        
+        # Create workflow info and engine
+        workflow_info = WorkflowInfo(
+            workflow_type=workflow_type_name,
+            workflow_domain=self._client.domain,
+            workflow_id=workflow_id,
+            workflow_run_id=run_id
+        )
 
-        # Create workflow context and execute with it active
-        workflow_engine = self._workflow_engines[engine_key]
+        self._workflow_engine = WorkflowEngine(
+            info=workflow_info, 
+            client=self._client, 
+            workflow_func=workflow_func
+        )
 
-        decision_result = await workflow_engine.process_decision(task)
+        decision_result = await self._workflow_engine.process_decision(task)
 
         # Respond with the decisions
         await self._respond_decision_task_completed(task, decision_result)
@@ -102,21 +95,22 @@ async def handle_task_failure(self, task: PollForDecisionTaskResponse, error: Ex
             task: The task that failed
             error: The exception that occurred
         """
+        logger.error(f"Decision task failed: {error}")
+        
+        # Determine the failure cause
+        cause = DecisionTaskFailedCause.DECISION_TASK_FAILED_CAUSE_UNHANDLED_DECISION
+        if isinstance(error, KeyError):
+            cause = DecisionTaskFailedCause.DECISION_TASK_FAILED_CAUSE_WORKFLOW_WORKER_UNHANDLED_FAILURE
+        elif isinstance(error, ValueError):
+            cause = DecisionTaskFailedCause.DECISION_TASK_FAILED_CAUSE_BAD_SCHEDULE_ACTIVITY_ATTRIBUTES
+        
+        # Create error details
+        # TODO: Use a data converter for error details serialization
+        error_message = str(error).encode('utf-8')
+        details = Payload(data=error_message)
+        
+        # Respond with failure
         try:
-            logger.error(f"Decision task failed: {error}")
-            
-            # Determine the failure cause
-            cause = DecisionTaskFailedCause.DECISION_TASK_FAILED_CAUSE_UNHANDLED_DECISION
-            if isinstance(error, KeyError):
-                cause = DecisionTaskFailedCause.DECISION_TASK_FAILED_CAUSE_WORKFLOW_WORKER_UNHANDLED_FAILURE
-            elif isinstance(error, ValueError):
-                cause = DecisionTaskFailedCause.DECISION_TASK_FAILED_CAUSE_BAD_SCHEDULE_ACTIVITY_ATTRIBUTES
-            
-            # Create error details
-            error_message = str(error).encode('utf-8')
-            details = Payload(data=error_message)
-            
-            # Respond with failure
             await self._client.worker_stub.RespondDecisionTaskFailed(
                 RespondDecisionTaskFailedRequest(
                     task_token=task.task_token,
@@ -125,11 +119,10 @@ async def handle_task_failure(self, task: PollForDecisionTaskResponse, error: Ex
                     details=details
                 )
             )
-            
             logger.info("Decision task failure response sent")
-            
         except Exception:
-            logger.exception("Error handling decision task failure")
+            logger.exception("Error responding to decision task failure")
+            
 
     async def _respond_decision_task_completed(self, task: PollForDecisionTaskResponse, decision_result: DecisionResult) -> None:
         """
@@ -144,30 +137,13 @@ async def _respond_decision_task_completed(self, task: PollForDecisionTaskRespon
                 task_token=task.task_token,
                 decisions=decision_result.decisions,
                 identity=self._identity,
-                return_new_decision_task=decision_result.force_create_new_decision_task,
-                force_create_new_decision_task=decision_result.force_create_new_decision_task
+                return_new_decision_task=True,
+                force_create_new_decision_task=False
             )
 
-            # Add query results if present
-            if decision_result.query_results:
-                request.query_results.update(decision_result.query_results)
-            
             await self._client.worker_stub.RespondDecisionTaskCompleted(request)
             logger.debug(f"Decision task completed with {len(decision_result.decisions)} decisions")
 
         except Exception:
             logger.exception("Error responding to decision task completion")
             raise
-    
-    def cleanup_workflow_engine(self, workflow_id: str, run_id: str) -> None:
-        """
-        Clean up a workflow engine when workflow execution is complete.
-        
-        Args:
-            workflow_id: The workflow ID
-            run_id: The run ID
-        """
-        engine_key = f"{workflow_id}:{run_id}"
-        if engine_key in self._workflow_engines:
-            del self._workflow_engines[engine_key]
-            logger.debug(f"Cleaned up workflow engine for {workflow_id}:{run_id}")

tests/cadence/_internal/test_decision_state_machine.py

@@ -439,7 +439,3 @@ def test_manager_aggregates_and_routes():
             ),
         )
     )
-
-    assert a.status is DecisionState.COMPLETED
-    assert t.status is DecisionState.COMPLETED
-    assert c.status is DecisionState.COMPLETED

tests/cadence/worker/test_decision_task_handler.py

@@ -75,8 +75,6 @@ def test_initialization(self, mock_client, mock_registry):
         assert handler._identity == "test_identity"
         assert handler._registry == mock_registry
         assert handler._options == {"option1": "value1"}
-        assert isinstance(handler._workflow_engines, dict)
-        assert len(handler._workflow_engines) == 0
 
     @pytest.mark.asyncio
     async def test_handle_task_implementation_success(self, handler, sample_decision_task, mock_registry):
@@ -139,8 +137,8 @@ async def test_handle_task_implementation_workflow_not_found(self, handler, samp
             await handler._handle_task_implementation(sample_decision_task)
 
     @pytest.mark.asyncio
-    async def test_handle_task_implementation_reuses_existing_engine(self, handler, sample_decision_task, mock_registry):
-        """Test that decision task handler reuses existing workflow engine."""
+    async def test_handle_task_implementation_creates_new_engine(self, handler, sample_decision_task, mock_registry):
+        """Test that decision task handler creates new workflow engine for each task."""
         # Mock workflow function
         mock_workflow_func = Mock()
         mock_registry.get_workflow.return_value = mock_workflow_func
@@ -153,23 +151,19 @@ async def test_handle_task_implementation_reuses_existing_engine(self, handler,
         mock_decision_result.query_results = {}
         mock_engine.process_decision = AsyncMock(return_value=mock_decision_result)
 
-        with patch('cadence.worker._decision_task_handler.WorkflowEngine', return_value=mock_engine):
+        with patch('cadence.worker._decision_task_handler.WorkflowEngine', return_value=mock_engine) as mock_engine_class:
             # First call - should create new engine
             await handler._handle_task_implementation(sample_decision_task)
 
-            # Second call - should reuse existing engine
+            # Second call - should create another new engine
             await handler._handle_task_implementation(sample_decision_task)
 
-        # Registry should only be called once
-        mock_registry.get_workflow.assert_called_once_with("TestWorkflow")
+        # Registry should be called for each task
+        assert mock_registry.get_workflow.call_count == 2
 
-        # Engine should be called twice
+        # Engine should be created twice and called twice
+        assert mock_engine_class.call_count == 2
         assert mock_engine.process_decision.call_count == 2
-        
-        # Should have one engine in the cache
-        assert len(handler._workflow_engines) == 1
-        engine_key = "test_workflow_id:test_run_id"
-        assert engine_key in handler._workflow_engines
 
     @pytest.mark.asyncio
     async def test_handle_task_failure_keyerror(self, handler, sample_decision_task):
@@ -237,8 +231,6 @@ async def test_respond_decision_task_completed_success(self, handler, sample_dec
         """Test successful decision task completion response."""
         decision_result = Mock(spec=DecisionResult)
         decision_result.decisions = [Decision(), Decision()]
-        decision_result.force_create_new_decision_task = True
-        decision_result.query_results = None  # Test without query results first
 
         await handler._respond_decision_task_completed(sample_decision_task, decision_result)
 
@@ -248,62 +240,33 @@ async def test_respond_decision_task_completed_success(self, handler, sample_dec
         assert call_args.task_token == sample_decision_task.task_token
         assert call_args.identity == handler._identity
         assert call_args.return_new_decision_task
-        assert call_args.force_create_new_decision_task
+        assert not call_args.force_create_new_decision_task
         assert len(call_args.decisions) == 2
-        # query_results should not be set when None
-        assert not hasattr(call_args, 'query_results') or len(call_args.query_results) == 0
 
     @pytest.mark.asyncio
     async def test_respond_decision_task_completed_no_query_results(self, handler, sample_decision_task):
         """Test decision task completion response without query results."""
         decision_result = Mock(spec=DecisionResult)
         decision_result.decisions = []
-        decision_result.force_create_new_decision_task = False
-        decision_result.query_results = None
 
         await handler._respond_decision_task_completed(sample_decision_task, decision_result)
 
         call_args = handler._client.worker_stub.RespondDecisionTaskCompleted.call_args[0][0]
-        assert not call_args.return_new_decision_task
+        assert call_args.return_new_decision_task
         assert not call_args.force_create_new_decision_task
         assert len(call_args.decisions) == 0
-        # query_results should not be set when None
-        assert not hasattr(call_args, 'query_results') or len(call_args.query_results) == 0
 
     @pytest.mark.asyncio
     async def test_respond_decision_task_completed_error(self, handler, sample_decision_task):
         """Test decision task completion response error handling."""
         decision_result = Mock(spec=DecisionResult)
         decision_result.decisions = []
-        decision_result.force_create_new_decision_task = False
-        decision_result.query_results = {}
 
         handler._client.worker_stub.RespondDecisionTaskCompleted.side_effect = Exception("Respond failed")
 
         with pytest.raises(Exception, match="Respond failed"):
             await handler._respond_decision_task_completed(sample_decision_task, decision_result)
 
-    def test_cleanup_workflow_engine(self, handler):
-        """Test workflow engine cleanup."""
-        # Add some mock engines
-        handler._workflow_engines["workflow1:run1"] = Mock()
-        handler._workflow_engines["workflow2:run2"] = Mock()
-        
-        # Clean up one engine
-        handler.cleanup_workflow_engine("workflow1", "run1")
-        
-        # Verify only one engine was removed
-        assert len(handler._workflow_engines) == 1
-        assert "workflow1:run1" not in handler._workflow_engines
-        assert "workflow2:run2" in handler._workflow_engines
-    
-    def test_cleanup_workflow_engine_not_found(self, handler):
-        """Test cleanup of non-existent workflow engine."""
-        # Should not raise error
-        handler.cleanup_workflow_engine("nonexistent", "run")
-        
-        # Should not affect existing engines
-        assert len(handler._workflow_engines) == 0
 
     @pytest.mark.asyncio
     async def test_workflow_engine_creation_with_workflow_info(self, handler, sample_decision_task, mock_registry):