fix(aci): update workflow filter evaluation with workflow env (#87678)

Author: cathteng

src/sentry/workflow_engine/processors/workflow.py

@@ -1,5 +1,5 @@
 import logging
-from dataclasses import asdict
+from dataclasses import asdict, replace
 from enum import StrEnum
 
 import sentry_sdk
@@ -84,27 +84,38 @@ def evaluate_workflows_action_filters(
 ) -> BaseQuerySet[Action]:
     filtered_action_groups: set[DataConditionGroup] = set()
 
-    # gets the list of the workflow ids, and then get the workflow_data_condition_groups for those workflows
-    workflow_ids = {workflow.id for workflow in workflows}
+    # Gets the list of the workflow ids, and then get the workflow_data_condition_groups for those workflows
+    workflow_ids_to_envs = {workflow.id: workflow.environment for workflow in workflows}
 
-    action_conditions = DataConditionGroup.objects.filter(
-        workflowdataconditiongroup__workflow_id__in=workflow_ids
-    ).distinct()
+    action_conditions = (
+        DataConditionGroup.objects.filter(
+            workflowdataconditiongroup__workflow_id__in=list(workflow_ids_to_envs.keys())
+        )
+        .prefetch_related("workflowdataconditiongroup_set")
+        .distinct()
+    )
 
     for action_condition in action_conditions:
-        # TODO(cathy): attach correct workflow to job
+        workflow_job = job
+
+        workflow_data_condition_group = action_condition.workflowdataconditiongroup_set.first()
+
+        # Populate the workflow_env in the job for the action_condition evaluation
+        if workflow_data_condition_group:
+            workflow_job = replace(
+                job, workflow_env=workflow_data_condition_group.workflow.environment
+            )
 
         (evaluation, result), remaining_conditions = process_data_condition_group(
-            action_condition.id, job
+            action_condition.id, workflow_job
         )
 
         if remaining_conditions:
             # If there are remaining conditions for the action filter to evaluate,
             # then return the list of conditions to enqueue
-            condition_group = action_condition.workflowdataconditiongroup_set.first()
-            if condition_group:
+            if workflow_data_condition_group:
                 enqueue_workflow(
-                    condition_group.workflow,
+                    workflow_data_condition_group.workflow,
                     remaining_conditions,
                     job.event,
                     WorkflowDataConditionGroupType.ACTION_FILTER,

tests/sentry/workflow_engine/processors/test_workflow.py

@@ -108,6 +108,55 @@ def test_error_event__logger(self, mock_logger):
             },
         )
 
+    @patch("sentry.workflow_engine.processors.workflow.filter_recently_fired_workflow_actions")
+    def test_populate_workflow_env_for_filters(self, mock_filter):
+        # this should not pass because the environment is not None
+        self.error_workflow.update(environment=self.group_event.get_environment())
+        error_workflow_filters = self.create_data_condition_group(
+            logic_type=DataConditionGroup.Type.ANY_SHORT_CIRCUIT
+        )
+        self.create_data_condition(
+            condition_group=error_workflow_filters,
+            type=Condition.FIRST_SEEN_EVENT,
+            comparison=True,
+            condition_result=True,
+        )
+        self.create_workflow_data_condition_group(
+            workflow=self.error_workflow, condition_group=error_workflow_filters
+        )
+
+        workflow_triggers = self.create_data_condition_group(
+            logic_type=DataConditionGroup.Type.ANY_SHORT_CIRCUIT
+        )
+        workflow_filters = self.create_data_condition_group(
+            logic_type=DataConditionGroup.Type.ANY_SHORT_CIRCUIT
+        )
+        # this should pass because the environment is None
+        self.create_data_condition(
+            condition_group=workflow_filters,
+            type=Condition.FIRST_SEEN_EVENT,
+            comparison=True,
+            condition_result=True,
+        )
+        workflow = self.create_workflow(
+            name="testy",
+            when_condition_group=workflow_triggers,
+        )
+        self.create_detector_workflow(
+            detector=self.error_detector,
+            workflow=workflow,
+        )
+        self.create_workflow_data_condition_group(
+            workflow=workflow, condition_group=workflow_filters
+        )
+
+        assert self.job.group_state
+        self.job.group_state["is_new"] = True
+
+        process_workflows(self.job)
+
+        mock_filter.assert_called_with({workflow_filters}, self.group)
+
     def test_same_environment_only(self):
         env = self.create_environment(project=self.project)
         other_env = self.create_environment(project=self.project)