feat(workflow_engine): Hook up activity updates to process_workflows (#94678)

saponifi3d · web-flow · commit 76f649b31809 · 2025-07-07T13:47:57.000-07:00
## Description
Complete support for `Activity` updates from the issue platform. This
will filter to known activities that we can handle (right now, it's just
issue resolution).

- Needed to fix a few areas of logic to support the full end to end of
`process_workflows` with an activity update.
- Added Tests to ensure we could create an activity, run the task and it
will fetch any associated data an evaluate as exected.
diff --git a/src/sentry/workflow_engine/processors/workflow.py b/src/sentry/workflow_engine/processors/workflow.py
@@ -11,6 +11,7 @@
 
 from sentry import buffer, features
 from sentry.eventstore.models import GroupEvent
+from sentry.models.activity import Activity
 from sentry.models.environment import Environment
 from sentry.utils import json
 from sentry.workflow_engine.models import (
@@ -182,7 +183,7 @@ def evaluate_workflow_triggers(
             "group_id": event_data.group.id,
             "event_id": event_id,
             "event_data": asdict(event_data),
-            "event_environment_id": environment.id,
+            "event_environment_id": environment.id if environment else None,
             "triggered_workflows": [workflow.id for workflow in triggered_workflows],
         },
     )
@@ -271,7 +272,7 @@ def evaluate_workflows_action_filters(
     return filtered_action_groups
 
 
-def get_environment_by_event(event_data: WorkflowEventData) -> Environment:
+def get_environment_by_event(event_data: WorkflowEventData) -> Environment | None:
     if isinstance(event_data.event, GroupEvent):
         try:
             environment = event_data.event.get_environment()
@@ -283,22 +284,27 @@ def get_environment_by_event(event_data: WorkflowEventData) -> Environment:
             raise Environment.DoesNotExist("Environment does not exist for the event")
 
         return environment
+    elif isinstance(event_data.event, Activity):
+        return None
 
-    raise TypeError(
-        "Expected event_data.event to be an instance of GroupEvent, got %s" % type(event_data.event)
-    )
+    raise TypeError(f"Cannot access the environment from, {type(event_data.event)}.")
 
 
 def _get_associated_workflows(
-    detector: Detector, environment: Environment, event_data: WorkflowEventData
+    detector: Detector, environment: Environment | None, event_data: WorkflowEventData
 ) -> set[Workflow]:
     """
     This is a wrapper method to get the workflows associated with a detector and environment.
     Used in process_workflows to wrap the query + logging into a single method
     """
+    environment_filter = (
+        (Q(environment_id=None) | Q(environment_id=environment.id))
+        if environment
+        else Q(environment_id=None)
+    )
     workflows = set(
         Workflow.objects.filter(
-            (Q(environment_id=None) | Q(environment_id=environment.id)),
+            environment_filter,
             detectorworkflow__detector_id=detector.id,
             enabled=True,
         )
@@ -324,7 +330,7 @@ def _get_associated_workflows(
                 "group_id": event_data.group.id,
                 "event_id": event_id,
                 "event_data": asdict(event_data),
-                "event_environment_id": environment.id,
+                "event_environment_id": environment.id if environment else None,
                 "workflows": [workflow.id for workflow in workflows],
                 "detector_type": detector.type,
             },
@@ -335,7 +341,7 @@ def _get_associated_workflows(
 
 @log_context.root()
 def process_workflows(
-    event_data: WorkflowEventData, detector_id: DetectorId = None
+    event_data: WorkflowEventData, detector: Detector | None = None
 ) -> set[Workflow]:
     """
     This method will get the detector based on the event, and then gather the associated workflows.
@@ -345,13 +351,11 @@ def process_workflows(
     Finally, each of the triggered workflows will have their actions evaluated and executed.
     """
     try:
-        detector: Detector
-        if detector_id is not None:
-            detector = Detector.objects.get(id=detector_id)
-        elif isinstance(event_data.event, GroupEvent):
+        if detector is None and isinstance(event_data.event, GroupEvent):
             detector = get_detector_by_event(event_data)
-        else:
-            raise ValueError("Unable to determine the detector_id for the event")
+
+        if detector is None:
+            raise ValueError("Unable to determine the detector for the event")
 
         log_context.add_extras(detector_id=detector.id)
         organization = detector.project.organization
@@ -397,9 +401,11 @@ def process_workflows(
 
     actions_to_trigger = evaluate_workflows_action_filters(triggered_workflows, event_data)
     actions = filter_recently_fired_workflow_actions(actions_to_trigger, event_data)
+
     if not actions:
         # If there aren't any actions on the associated workflows, there's nothing to trigger
         return triggered_workflows
+
     create_workflow_fire_histories(detector, actions, event_data)
 
     with sentry_sdk.start_span(op="workflow_engine.process_workflows.trigger_actions"):
diff --git a/src/sentry/workflow_engine/tasks.py b/src/sentry/workflow_engine/tasks.py
@@ -1,6 +1,7 @@
+from django.db import router, transaction
 from google.api_core.exceptions import DeadlineExceeded, RetryError, ServiceUnavailable
 
-from sentry import nodestore
+from sentry import features, nodestore
 from sentry.eventstore.models import Event, GroupEvent
 from sentry.eventstream.base import GroupState
 from sentry.issues.issue_occurrence import IssueOccurrence
@@ -15,6 +16,7 @@
 from sentry.types.activity import ActivityType
 from sentry.utils import metrics
 from sentry.utils.retries import ConditionalRetryPolicy, exponential_delay
+from sentry.workflow_engine.models import Detector
 from sentry.workflow_engine.processors.workflow import process_workflows
 from sentry.workflow_engine.types import WorkflowEventData
 from sentry.workflow_engine.utils import log_context
@@ -42,17 +44,39 @@
         ),
     ),
 )
-def process_workflow_activity(activity_id: int, detector_id: int) -> None:
+def process_workflow_activity(activity_id: int, group_id: int, detector_id: int) -> None:
     """
-    Process a workflow task identified by the given Activity ID and Detector ID.
+    Process a workflow task identified by the given activity, group, and detector.
 
     The task will get the Activity from the database, create a WorkflowEventData object,
     and then process the data in `process_workflows`.
     """
-    # TODO - @saponifi3d - implement this in a follow-up PR. This update will require WorkflowEventData
-    # to allow for an activity in the `event` attribute. That refactor is a bit noisy
-    # and will be done in a subsequent pr.
-    pass
+    with transaction.atomic(router.db_for_write(Detector)):
+        try:
+            activity = Activity.objects.get(id=activity_id)
+            group = Group.objects.get(id=group_id)
+            detector = Detector.objects.get(id=detector_id)
+        except (Activity.DoesNotExist, Group.DoesNotExist, Detector.DoesNotExist):
+            logger.exception(
+                "Unable to fetch data to process workflow activity",
+                extra={
+                    "activity_id": activity_id,
+                    "group_id": group_id,
+                    "detector_id": detector_id,
+                },
+            )
+            return  # Exit execution that we cannot recover from
+
+    event_data = WorkflowEventData(
+        event=activity,
+        group=group,
+    )
+
+    process_workflows(event_data, detector)
+    metrics.incr(
+        "workflow_engine.process_workflow.activity_update.executed",
+        tags={"activity_type": activity.type},
+    )
 
 
 @group_status_update_registry.register("workflow_status_update")
@@ -63,8 +87,10 @@ def workflow_status_update_handler(
     Hook the process_workflow_task into the activity creation registry.
 
     Since this handler is called in process for the activity, we want
-    to queue a task to process workflows asynchronously.
-    """
+    to queue a task to process workflows asynchronously."""
+    metrics.incr(
+        "workflow_engine.process_workflow.activity_update", tags={"activity_type": activity.type}
+    )
     if activity.type not in SUPPORTED_ACTIVITIES:
         # If the activity type is not supported, we do not need to process it.
         return
@@ -77,11 +103,12 @@ def workflow_status_update_handler(
         metrics.incr("workflow_engine.error.tasks.no_detector_id")
         return
 
-    # TODO - implement in follow-up PR for now, just track a metric that we are seeing the activities.
-    # process_workflow_task.delay(activity.id, detector_id)
-    metrics.incr(
-        "workflow_engine.process_workflow.activity_update", tags={"activity_type": activity.type}
-    )
+    if features.has("organizations:workflow-engine-process-activity", group.organization):
+        process_workflow_activity.delay(
+            activity_id=activity.id,
+            group_id=group.id,
+            detector_id=detector_id,
+        )
 
 
 def _should_retry_nodestore_fetch(attempt: int, e: Exception) -> bool:
diff --git a/tests/sentry/workflow_engine/test_task.py b/tests/sentry/workflow_engine/test_task.py
@@ -6,8 +6,14 @@
 from sentry.models.activity import Activity
 from sentry.models.group import GroupStatus
 from sentry.testutils.cases import TestCase
+from sentry.testutils.helpers import with_feature
 from sentry.types.activity import ActivityType
-from sentry.workflow_engine.tasks import fetch_event, workflow_status_update_handler
+from sentry.workflow_engine.tasks import (
+    fetch_event,
+    process_workflow_activity,
+    workflow_status_update_handler,
+)
+from sentry.workflow_engine.types import WorkflowEventData
 
 
 class FetchEventTests(TestCase):
@@ -57,3 +63,142 @@ def test__no_detector_id(self):
         with mock.patch("sentry.workflow_engine.tasks.metrics.incr") as mock_incr:
             workflow_status_update_handler(group, message, activity)
             mock_incr.assert_called_with("workflow_engine.error.tasks.no_detector_id")
+
+    def test__feature_flag(self):
+        detector = self.create_detector(project=self.project)
+        group = self.create_group(project=self.project)
+        activity = Activity(
+            project=self.project,
+            group=group,
+            type=ActivityType.SET_RESOLVED.value,
+            data={"fingerprint": ["test_fingerprint"]},
+        )
+        message = StatusChangeMessageData(
+            id="test_message_id",
+            project_id=self.project.id,
+            new_status=GroupStatus.RESOLVED,
+            new_substatus=None,
+            fingerprint=["test_fingerprint"],
+            detector_id=detector.id,
+        )
+
+        with mock.patch(
+            "sentry.workflow_engine.tasks.process_workflow_activity.delay"
+        ) as mock_delay:
+            workflow_status_update_handler(group, message, activity)
+            mock_delay.assert_not_called()
+
+    @with_feature("organizations:workflow-engine-process-activity")
+    def test(self):
+        detector = self.create_detector(project=self.project)
+        group = self.create_group(project=self.project)
+        activity = Activity(
+            project=self.project,
+            group=group,
+            type=ActivityType.SET_RESOLVED.value,
+            data={"fingerprint": ["test_fingerprint"]},
+        )
+        message = StatusChangeMessageData(
+            id="test_message_id",
+            project_id=self.project.id,
+            new_status=GroupStatus.RESOLVED,
+            new_substatus=None,
+            fingerprint=["test_fingerprint"],
+            detector_id=detector.id,
+        )
+
+        with mock.patch(
+            "sentry.workflow_engine.tasks.process_workflow_activity.delay"
+        ) as mock_delay:
+            workflow_status_update_handler(group, message, activity)
+            mock_delay.assert_called_once_with(
+                activity_id=activity.id,
+                group_id=group.id,
+                detector_id=detector.id,
+            )
+
+
+class TestProcessWorkflowActivity(TestCase):
+    def setUp(self):
+        self.group = self.create_group(project=self.project)
+        self.activity = Activity(
+            project=self.project,
+            group=self.group,
+            type=ActivityType.SET_RESOLVED.value,
+            data={"fingerprint": ["test_fingerprint"]},
+        )
+        self.activity.save()
+        self.detector = self.create_detector()
+
+    def test_process_workflow_activity__no_workflows(self):
+        with mock.patch(
+            "sentry.workflow_engine.processors.workflow.evaluate_workflow_triggers",
+            return_value=set(),
+        ) as mock_evaluate:
+            process_workflow_activity.run(
+                activity_id=self.activity.id,
+                group_id=self.group.id,
+                detector_id=self.detector.id,
+            )
+            # Short-circuit evaluation, no workflows associated
+            assert mock_evaluate.call_count == 0
+
+    @mock.patch(
+        "sentry.workflow_engine.processors.workflow.evaluate_workflow_triggers", return_value=set()
+    )
+    @mock.patch(
+        "sentry.workflow_engine.processors.workflow.evaluate_workflows_action_filters",
+        return_value=set(),
+    )
+    def test_process_workflow_activity__workflows__no_actions(
+        self, mock_eval_actions, mock_evaluate
+    ):
+        self.workflow = self.create_workflow(organization=self.organization)
+        self.create_detector_workflow(
+            detector=self.detector,
+            workflow=self.workflow,
+        )
+
+        process_workflow_activity.run(
+            activity_id=self.activity.id,
+            group_id=self.group.id,
+            detector_id=self.detector.id,
+        )
+
+        event_data = WorkflowEventData(
+            event=self.activity,
+            group=self.group,
+        )
+
+        mock_evaluate.assert_called_once_with({self.workflow}, event_data)
+        assert mock_eval_actions.call_count == 0
+
+    @mock.patch("sentry.workflow_engine.processors.workflow.filter_recently_fired_workflow_actions")
+    def test_process_workflow_activity(self, mock_filter_actions):
+        self.workflow = self.create_workflow(organization=self.organization)
+
+        self.action_group = self.create_data_condition_group(logic_type="any-short")
+        self.action = self.create_action()
+        self.create_data_condition_group_action(
+            condition_group=self.action_group,
+            action=self.action,
+        )
+        self.create_workflow_data_condition_group(self.workflow, self.action_group)
+
+        self.create_detector_workflow(
+            detector=self.detector,
+            workflow=self.workflow,
+        )
+
+        expected_event_data = WorkflowEventData(
+            event=self.activity,
+            group=self.group,
+        )
+
+        process_workflow_activity.run(
+            activity_id=self.activity.id,
+            group_id=self.group.id,
+            detector_id=self.detector.id,
+        )
+
+        mock_filter_actions.assert_called_once_with({self.action_group}, expected_event_data)
diff --git a/tests/sentry/workflow_engine/test_task_integration.py b/tests/sentry/workflow_engine/test_task_integration.py
@@ -44,7 +44,7 @@ def test_handler_invoked__when_update_status_called(self):
         with mock.patch("sentry.workflow_engine.tasks.metrics.incr") as mock_incr:
             _process_message(message)
 
-            mock_incr.assert_called_with(
+            mock_incr.assert_any_call(
                 "workflow_engine.process_workflow.activity_update",
                 tags={"activity_type": ActivityType.SET_RESOLVED.value},
             )
@@ -66,7 +66,7 @@ def test_handler_invoked__when_resolved(self):
 
         with mock.patch("sentry.workflow_engine.tasks.metrics.incr") as mock_incr:
             update_status(self.group, message)
-            mock_incr.assert_called_with(
+            mock_incr.assert_any_call(
                 "workflow_engine.process_workflow.activity_update",
                 tags={"activity_type": ActivityType.SET_RESOLVED.value},
             )
